Skip to main content
Version: Next

AmazonSqs

Amazon SQS sink connector

Description​

The Amazon SQS sink connector writes each incoming SeaTunnel row to one Amazon SQS queue URL. The row is serialized by the configured format, and the serialized value is sent as the SQS message body.

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Key Features​

Sink Options​

NameTypeRequiredDefaultDescription
urlStringYes-Full SQS queue URL to write to, for example https://sqs.us-east-1.amazonaws.com/123456789012/sink_queue.
regionStringYes-AWS region of the SQS queue, for example us-east-1.
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.
common-optionsNo-Sink plugin common parameters. For details, see Sink Common Options.

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

Format Notes​

  • json writes each row as a JSON object.
  • text joins row fields by field_delimiter.
  • canal_json writes Canal JSON messages. For details, see Canal JSON.
  • debezium_json writes Debezium JSON messages. For details, see Debezium JSON.
  • The sink sends only the message body. It does not expose SQS message attributes, delay seconds, deduplication ID, or message group ID options.
  • access_key_id and secret_access_key are optional, but they must be configured together when static AWS credentials are used.
  • The sink sends each SeaTunnel row as one SQS message. It does not batch multiple rows into one SQS request.

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"
  }
}

Write JSON Messages​

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

source {
  FakeSource {
    row.num = 1
    schema = {
      fields {
        name = string
      }
    }
    rows = [
      {
        kind = INSERT
        fields = ["test_name"]
      }
    ]
  }
}

sink {
  AmazonSqs {
    url = "https://sqs.us-east-1.amazonaws.com/123456789012/sink_queue"
    region = "us-east-1"
    access_key_id = "AKIA..."
    secret_access_key = "SECRET..."
  }
}

Write Text Messages With a Custom Delimiter​

source {
  FakeSource {
    schema = {
      fields {
        artist = string
        album = string
        release_year = int
      }
    }
  }
}

sink {
  AmazonSqs {
    url = "https://sqs.us-east-1.amazonaws.com/123456789012/sink_queue"
    region = "us-east-1"
    format = text
    field_delimiter = "|"
  }
}

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