跳到主要内容
版本：Next

Tablestore

Tablestore Sink 连接器

描述​

将 SeaTunnel 数据写入阿里云 Tablestore。

主要特性​

选项​

参数名类型是否必填默认值描述
end_pointstring是-Tablestore 访问地址。
instance_namestring是-Tablestore 实例名称。
access_key_idstring是-访问 Tablestore 使用的 AccessKey ID。
access_key_secretstring是-访问 Tablestore 使用的 AccessKey Secret。
tablestring是-目标 Tablestore 表名。
primary_keysarray是-目标表的主键字段名。
schemaconfig是-输入数据结构。主键字段也必须包含在 schema.fields 中。
batch_sizeint否25单次批量写入的最大数据条数。
common-optionsconfig否-Sink 通用选项。

使用说明​

  • primary_keys 可以包含一个或多个主键字段。这些字段会写为 Tablestore 主键列，其余字段会写为普通属性列。
  • Sink 使用 Tablestore RowPutChange 写入，并使用 RowExistenceExpectation.IGNORE。当上游发送 DELETE 类型数据时，当前 Sink 不会删除 Tablestore 中的行。
  • batch_size 控制缓存多少行后刷新；任务关闭时，写入器也会刷新剩余数据。
  • 不建议把 access_key_id 和 access_key_secret 直接写入会提交到代码仓库的任务文件中，建议使用运行时变量替换或部署环境支持的密钥管理方式。

common options [config]​

Sink 插件通用参数，请参考 Sink Common Options。

示例​

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

source {
  FakeSource {
    row.num = 2
    schema = {
      fields {
        order_id = string
        user_id = string
        amount = double
      }
    }
    rows = [
      {
        fields = ["order-1", "user-1", 99.5]
      },
      {
        fields = ["order-2", "user-2", 20.0]
      }
    ]
  }
}

sink {
  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"]
    batch_size = 25
    schema = {
      fields {
        order_id = string
        user_id = string
        amount = double
      }
    }
  }
}

类型映射​

SeaTunnel 类型Tablestore 普通属性列类型Tablestore 主键列类型
INT, TINYINT, SMALLINT, BIGINTINTEGERINTEGER
FLOAT, DOUBLE, DECIMALDOUBLESTRING
STRING, DATE, TIME, TIMESTAMPSTRINGSTRING
BOOLEANBOOLEANSTRING
BYTESBINARYBINARY

变更日志​

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