跳到主要内容
版本：Next

AmazonSqs

Amazon SQS 源连接器

描述​

Amazon SQS 源连接器用于从一个 Amazon SQS 队列 URL 读取消息。连接器会按照 format 和 schema 解析每条消息的消息体，然后输出为 SeaTunnel 行数据。

该连接器使用单个 reader，处理完本次接收到的消息后任务会结束，适合从队列中做有界读取。

每次 receive 请求最多从 SQS 拉取 10 条消息。如果队列里还有更多消息，需要再次运行任务，或者使用上游调度方式重复触发这种有界读取。

支持的引擎​

Spark
Flink
SeaTunnel Zeta

主要特性​

源选项​

名称类型是否必填默认值描述
urlString是-要读取的完整 SQS 队列 URL，例如 https://sqs.us-east-1.amazonaws.com/123456789012/source_queue。
regionString是-SQS 队列所在的 AWS 区域，例如 us-east-1。
schemaConfig是-消息体结构，包含字段名和字段类型。更多说明请参考 Schema 特性。
access_key_idString否-AWS access key ID。和 secret_access_key 一起配置时使用静态凭证；两者都不配置时使用 AWS 默认凭证链。
secret_access_keyString否-AWS secret access key。和 access_key_id 一起配置时使用静态凭证。
formatString否json消息体格式。支持 json、text、canal_json、debezium_json。
field_delimiterString否,当 format = text 时使用的字段分隔符。
delete_messageBoolean否false读取并成功解析消息后，是否从队列中删除该消息。
message_group_idString否-为兼容保留的消息分组 ID 选项，普通 SQS 读取不需要配置。
debezium_record_include_schemaBoolean否trueDebezium JSON 消息是否包含 schema。仅在 format = debezium_json 时使用。
common-options否-源插件通用参数，详见 Source Common Options。

url 可以指向 AWS SQS，也可以指向兼容 SQS 的本地服务，例如 http://sqs-host:4566/000000000000/source_queue。

格式说明​

  • json：把每条消息体按 JSON 对象解析，并要求字段能对应到 schema。
  • text：按 field_delimiter 切分消息体，并按 schema 中字段顺序映射。
  • canal_json：读取 Canal JSON 消息，详见 Canal JSON。
  • debezium_json：读取 Debezium JSON 消息，详见 Debezium JSON。
  • delete_message = true 会删除已经消费的 SQS 消息。如果只是检查或复制消息，建议保留默认值 false。
  • access_key_id 和 secret_access_key 是可选项；如果使用静态 AWS 凭证，需要两个一起配置。
  • 该源连接器只执行一次 receive 请求，最多读取 10 条消息，然后结束这个有界任务。

任务示例​

在本地兼容队列之间复制消息​

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

读取 JSON 消息​

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

使用自定义分隔符读取文本消息​

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

变更日志​