Skip to main content
Version: Next

AmazonSqs

Amazon SQS source connector

Description​

The Amazon SQS source connector reads messages from one Amazon SQS queue URL. Each message body is deserialized by the configured format and schema, then emitted as SeaTunnel rows.

The connector uses a single reader and finishes the job after the current receive request is processed. It is suitable for bounded reads from a queue.

Each receive request asks SQS for up to 10 messages. If more messages are waiting in the queue, run another job or use a streaming-style upstream design that repeatedly starts bounded reads.

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Key Features​

Source Options​

NameTypeRequiredDefaultDescription
urlStringYes-Full SQS queue URL to read from, for example https://sqs.us-east-1.amazonaws.com/123456789012/source_queue.
regionStringYes-AWS region of the SQS queue, for example us-east-1.
schemaConfigYes-Message body schema. For more details, see Schema Feature.
access_key_idStringNo-AWS access key ID. Set it together with secret_access_key to use static credentials. Leave both unset to use the AWS default credential provider chain.
secret_access_keyStringNo-AWS secret access key. Set it together with access_key_id to use static credentials.
formatStringNojsonMessage body format. Supported values are json, text, canal_json, and debezium_json.
field_delimiterStringNo,Field delimiter used when format = text.
delete_messageBooleanNofalseWhether to delete a message from the queue after it is read and deserialized successfully.
message_group_idStringNo-Message group ID option kept for compatibility. It is not required for normal SQS reads.
debezium_record_include_schemaBooleanNotrueWhether Debezium JSON messages include a schema. This option is used only when format = debezium_json.
common-optionsNo-Source plugin common parameters. For details, see Source Common Options.

url can point to AWS SQS or to an SQS-compatible local service, for example http://sqs-host:4566/000000000000/source_queue.

Format Notes​

  • json reads each message body as a JSON object that matches schema.
  • text splits each message body by field_delimiter and maps the values to fields in schema order.
  • canal_json reads Canal JSON messages. For details, see Canal JSON.
  • debezium_json reads Debezium JSON messages. For details, see Debezium JSON.
  • delete_message = true removes consumed messages from SQS. Keep the default false when you only want to inspect or copy messages without deleting them.
  • access_key_id and secret_access_key are optional, but they must be configured together when static AWS credentials are used.
  • The source performs one receive request, with up to 10 messages, and then finishes the bounded job.

Task Examples​

Copy Messages Between Local-Compatible Queues​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  AmazonSqs {
    url = "http://sqs-host:4566/000000000000/source_queue"
    access_key_id = "1234"
    secret_access_key = "abcd"
    region = "us-east-1"
    schema = {
      fields {
        name = "string"
      }
    }
  }
}

sink {
  AmazonSqs {
    url = "http://sqs-host:4566/000000000000/sink_queue"
    access_key_id = "1234"
    secret_access_key = "abcd"
    region = "us-east-1"
  }
}

Read JSON Messages​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  AmazonSqs {
    url = "https://sqs.us-east-1.amazonaws.com/123456789012/source_queue"
    region = "us-east-1"
    access_key_id = "AKIA..."
    secret_access_key = "SECRET..."
    schema = {
      fields {
        name = string
      }
    }
  }
}

sink {
  Console {}
}

Read Text Messages With a Custom Delimiter​

source {
  AmazonSqs {
    url = "https://sqs.us-east-1.amazonaws.com/123456789012/source_queue"
    region = "us-east-1"
    format = text
    field_delimiter = "#"
    delete_message = true
    schema = {
      fields {
        artist = string
        album = string
        release_year = int
      }
    }
  }
}

sink {
  Console {}
}

Changelog​

Change Log
ChangeCommitVersion
[Improve][Core] Unify the aws-sdk-v2 version to 2.31.30 (#9698)https://github.com/apache/seatunnel/commit/41c251cc8a2.3.12
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[improve] amazon sqs connector update (#8602)https://github.com/apache/seatunnel/commit/c747e02a982.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[Improve] Remove all useless prepare, getProducedType method (#5741)https://github.com/apache/seatunnel/commit/ed94fffbb92.3.4
[Improve][Connector-V2] Change amazonsqs to AmazonSqs as connector identifier (#5742)https://github.com/apache/seatunnel/commit/245705d0f72.3.4
[Feature][Connector-V2] Add connector amazonsqs (#5367)https://github.com/apache/seatunnel/commit/7f75a8eafd2.3.4