Tablestore

Tablestore sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Description

Write SeaTunnel rows to Alibaba Cloud Tablestore.

Key features

• exactly-once
• cdc
• support multiple table write
• timer flush

Data Type Mapping

SeaTunnel type	Tablestore attribute column type	Tablestore primary key type
INT, TINYINT, SMALLINT, BIGINT	INTEGER	INTEGER
FLOAT, DOUBLE, DECIMAL	DOUBLE	STRING
STRING, DATE, TIME, TIMESTAMP	STRING	STRING
BOOLEAN	BOOLEAN	STRING
BYTES	BINARY	BINARY

Options

name	type	required	default value	description
end_point	string	yes	-	Tablestore endpoint.
instance_name	string	yes	-	Tablestore instance name.
access_key_id	string	yes	-	AccessKey ID used to access Tablestore.
access_key_secret	string	yes	-	AccessKey secret used to access Tablestore.
table	string	yes	-	Target Tablestore table name.
primary_keys	array	yes	-	Primary key field names in the target table.
schema	config	yes	-	Input schema. Primary key fields must also exist in schema.fields.
batch_size	int	no	25	Maximum number of rows written in one batch.
common-options	config	no	-	Sink common options.

Usage notes

• primary_keys can contain one or more primary key fields. These fields are written as Tablestore primary key columns; all other schema fields are written as normal attribute columns.
• The sink writes rows with Tablestore RowPutChange and RowExistenceExpectation.IGNORE. It does not delete rows when upstream sends DELETE row kinds.
• batch_size controls when buffered rows are flushed. The writer also flushes remaining rows when the job closes.
• Keep access_key_id and access_key_secret out of committed job files. Prefer runtime variable substitution or a secret manager supported by your deployment environment.

common options [config]

Sink plugin common parameters, please refer to Sink Common Options for details.

Example

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

Changelog

Change Log

Change	Commit	Version
[Improve] table_store options (#9515)	https://github.com/apache/seatunnel/commit/145b68793f	2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Feature][Connector-V2][Tablestore] Support Source connector for Tablestore #7448 (#7467)	https://github.com/apache/seatunnel/commit/a7ca51b585	2.3.8
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Improve][Connector-V2] Remove scheduler in Tablestore sink (#5272)	https://github.com/apache/seatunnel/commit/8d6b07e466	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][TableStore] Unified excetion for TableStore sink connector (#3527)	https://github.com/apache/seatunnel/commit/7b264d7004	2.3.0
[Feature][connector-v2] add tablestore source and sink (#3309)	https://github.com/apache/seatunnel/commit/ebebf0b633	2.3.0