跳到主要内容
版本：Next

Tablestore

Tablestore 源连接器

描述​

从阿里云 Tablestore 读取全量和增量数据。该连接器使用 Tablestore Tunnel 的 BaseAndStream 模式，先读取已有数据，再继续消费后续变更。

主要特性​

选项​

参数名类型是否必填默认值描述
end_pointstring是-Tablestore 访问地址，例如 https://<instance>.<region>.ots.aliyuncs.com。
instance_namestring是-Tablestore 实例名称。
access_key_idstring是-访问 Tablestore 使用的 AccessKey ID。
access_key_secretstring是-访问 Tablestore 使用的 AccessKey Secret。
tablestring是-Tablestore 表名。读取多张表时，用英文逗号分隔。
primary_keysarray是-主键名。读取多张表时，需要按 table 的顺序为每张表配置一个主键名。
schemaconfig是-输出数据结构。更多详情请参考 Schema 特性。

使用说明​

  • job.mode = "BATCH" 会读取有界数据；job.mode = "STREAMING" 会在读取已有数据后继续消费增量记录。
  • 当 table 配置多张表时，primary_keys 的数量必须和表数量一致。例如 table = "orders,users" 且两张表都使用 id 作为主键字段时，可以配置 primary_keys = ["id", "id"]。
  • 多表读取共用一个 schema 配置，因此这些表的输出字段需要保持兼容。
  • 源连接器会根据 Tablestore 的变更记录输出 INSERT、UPDATE_AFTER 和 DELETE 类型的数据。
  • 不建议把 access_key_id 和 access_key_secret 直接写入会提交到代码仓库的任务文件中，建议使用运行时变量替换或部署环境支持的密钥管理方式。

示例​

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

source {
  Tablestore {
    end_point = "https://<instance>.<region>.ots.aliyuncs.com"
    instance_name = "<instance-name>"
    access_key_id = "${ACCESS_KEY_ID}"
    access_key_secret = "${ACCESS_KEY_SECRET}"
    table = "orders"
    primary_keys = ["order_id"]
    schema = {
      fields {
        order_id = string
        user_id = string
        amount = double
        updated_at = string
      }
    }
  }
}

sink {
  Console {}
}

多表示例​

env {
  parallelism = 2
  job.mode = "STREAMING"
}

source {
  Tablestore {
    end_point = "https://<instance>.<region>.ots.aliyuncs.com"
    instance_name = "<instance-name>"
    access_key_id = "${ACCESS_KEY_ID}"
    access_key_secret = "${ACCESS_KEY_SECRET}"
    table = "orders,users"
    primary_keys = ["id", "id"]
    schema = {
      fields {
        id = string
        value = string
        updated_at = string
      }
    }
  }
}

sink {
  Console {}
}

变更日志​

Change Log
ChangeCommitVersion
[Improve] table_store options (#9515)https://github.com/apache/seatunnel/commit/145b68793f2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.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][Connector-V2][Tablestore] Support Source connector for Tablestore #7448 (#7467)https://github.com/apache/seatunnel/commit/a7ca51b5852.3.8
[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
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Improve][Connector-V2] Remove scheduler in Tablestore sink (#5272)https://github.com/apache/seatunnel/commit/8d6b07e4662.3.3
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][TableStore] Unified excetion for TableStore sink connector (#3527)https://github.com/apache/seatunnel/commit/7b264d70042.3.0
[Feature][connector-v2] add tablestore source and sink (#3309)https://github.com/apache/seatunnel/commit/ebebf0b6332.3.0