This topic provides the DDL syntax that is used to create a Tablestore result table, describes the parameters in the WITH clause, and provides data type mappings.

What is Tablestore?

Tablestore is a distributed NoSQL database service that is built on the Apsara distributed operating system of Alibaba Cloud. Tablestore adopts sharding and load balancing technologies to seamlessly scale out services and handle concurrent transactions. You can use Tablestore to store and query large amounts of structured data in real time. Flink can use the streaming data that is stored in Tablestore as output data.

Prerequisites

A Tablestore data table is created. For more information, see Create tables.

Limits

Only Flink that uses Ververica Runtime (VVR) 2.0.0 or later supports Tablestore connectors.

DDL syntax

CREATE TABLE ots_sink (
  name VARCHAR,
  age BIGINT,
  birthday BIGINT,
  primary key(name,age) not enforced
) WITH (
  'connector'='ots',
  'instanceName'='<yourInstanceName>',
  'tableName'='<yourTableName>',
  'accessId'='<yourAccessId>',
  'accessKey'='<yourAccessSecret>',
  'endPoint'='<yourEndpoint>',
  'valueColumns'='birthday'
); 
Note You must define a primary key for a Tablestore result table. Latest output data is appended to the Tablestore result table to update the result.

Parameters in the WITH clause

Parameter Description Required Remarks
connector The type of the result table. Yes Set the value to ots.
instanceName The name of the Tablestore instance. Yes N/A.
tableName The name of the table. Yes N/A.
endPoint The endpoint of the Tablestore instance. Yes For more information, see Endpoint.
accessId The AccessKey ID that is used to access the Tablestore instance. Yes N/A.
accessKey The AccessKey secret that is used to access the Tablestore instance. Yes N/A.
valueColumns The names of the columns that you want to insert. Yes Separate multiple columns with commas (,), such as ID,NAME.
bufferSize The maximum number of data records that can be stored in the buffer before data deduplication is triggered. No Default value: 5000. This value indicates that data deduplication is triggered if the number of input data records in the buffer reaches 5,000.
batchWriteTimeoutMs The write timeout period. No Default value: 5000. Unit: milliseconds. This value indicates that if the number of input data records does not reach the value specified by the batchSize parameter within 5,000 milliseconds, all cached data is written to the result table.
batchSize The number of data records that are written at a time. No Default value: 100.
retryIntervalMs The retry interval. No Default value: 1000. Unit: milliseconds.
maxRetryTimes The maximum number of retries. No Default value: 10.
ignoreDelete Specifies whether to ignore delete operations. No Default value: False.
connectTimeout The timeout period for a connector to connect to Tablestore. No Default value: 30000. Unit: milliseconds.
socketTimeout The socket timeout period for a connector to connect to Tablestore. No Default value: 30000. Unit: milliseconds.

Data type mapping

Data type of Tablestore Data type of Flink
INTEGER BIGINT
STRING STRING
BOOLEAN BOOLEAN
DOUBLE DOUBLE