Realtime Compute for Apache Flink:Tablestore

General options

All table types share the following options.

Source table options

Sink table options

Dimension table options

How the cache works

The dimension table cache reduces repeated lookups against Tablestore. Choose a cache policy based on your table size and query patterns:

• None: No caching. Every lookup hits Tablestore directly. Use when data changes frequently and freshness is critical.

• LRU: Caches a fixed number of recently accessed records. When a lookup misses the cache, the connector queries Tablestore and updates the cache with the result. Set cacheSize and cacheTTLMs when using this policy.

• ALL (default): Loads the entire dimension table into the cache before the job starts. All subsequent lookups are served from cache. When the cache expires (cacheTTLMs), the connector reloads all data. Use ALL when the table is small and you expect many missing-key lookups. When using ALL, increase the memory of the join node — the cache requires approximately twice the size of the remote table.

Read from Tablestore and write to Tablestore

This example reads order data from a Tablestore source table via Tunnel Service and writes it to a Tablestore sink table. The sink table uses an auto-increment primary key column.

CREATE TEMPORARY TABLE tablestore_stream (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='flink-source',
  'tableName'='flink_source_table',
  'tunnelName'='flinksourcestream',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'ignoreDelete'='false',
  'skipInvalidData'='false'
);

CREATE TEMPORARY TABLE ots_sink (
  `order` VARCHAR,
  orderid VARCHAR,
  customerid VARCHAR,
  customername VARCHAR,
  PRIMARY KEY (`order`, orderid) NOT ENFORCED
) WITH (
  'connector'='ots',
  'endPoint'='<yourEndpoint>',
  'instanceName'='flink-sink',
  'tableName'='flink_sink_table',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'valueColumns'='customerid,customername',
  'autoIncrementKey'='${auto_increment_primary_key_name}'
);

INSERT INTO ots_sink
SELECT `order`, orderid, customerid, customername FROM tablestore_stream;

Synchronize a wide-column table to a time series table

This example reads from a wide-column source table and writes to a time series sink table. The sink table's tags column uses MAP<STRING, STRING> to hold tag key-value pairs, and storageType is set to TIMESERIES.

CREATE TEMPORARY TABLE timeseries_source (
  measurement STRING,
  datasource STRING,
  tag_a STRING,
  `time` BIGINT,
  binary_value BINARY,
  bool_value BOOLEAN,
  double_value DOUBLE,
  long_value BIGINT,
  string_value STRING,
  tag_b STRING,
  tag_c STRING,
  tag_d STRING,
  tag_e STRING,
  tag_f STRING
) WITH (
  'connector'='ots',
  'endPoint'='https://iotstore-test.cn-hangzhou.vpc.tablestore.aliyuncs.com',
  'instanceName'='iotstore-test',
  'tableName'='test_ots_timeseries_2',
  'tunnelName'='timeseries_source_tunnel_2',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'ignoreDelete'='true'
);

CREATE TEMPORARY TABLE timeseries_sink (
  measurement STRING,
  datasource STRING,
  tags MAP<STRING, STRING>,
  `time` BIGINT,
  binary_value BINARY,
  bool_value BOOLEAN,
  double_value DOUBLE,
  long_value BIGINT,
  string_value STRING,
  tag_b STRING,
  tag_c STRING,
  tag_d STRING,
  tag_e STRING,
  tag_f STRING,
  PRIMARY KEY (measurement, datasource, tags, `time`) NOT ENFORCED
) WITH (
  'connector'='ots',
  'endPoint'='https://iotstore-test.cn-hangzhou.vpc.tablestore.aliyuncs.com',
  'instanceName'='iotstore-test',
  'tableName'='test_timeseries_sink_table_2',
  'accessId'='${ak_id}',
  'accessKey'='${ak_secret}',
  'storageType'='TIMESERIES'
);

-- Build the tags map from individual tag columns and insert into the time series sink table
INSERT INTO timeseries_sink
SELECT
  measurement,
  datasource,
  MAP['tag_a', tag_a, 'tag_b', tag_b, 'tag_c', tag_c, 'tag_d', tag_d, 'tag_e', tag_e, 'tag_f', tag_f] AS tags,
  `time`,
  binary_value,
  bool_value,
  double_value,
  long_value,
  string_value,
  tag_b,
  tag_c,
  tag_d,
  tag_e,
  tag_f
FROM timeseries_source;