Realtime Compute for Apache Flink:Tablestore

共通オプション

以下のオプションは、すべてのテーブルタイプで共通です。

ソーステーブルのオプション

シンクテーブルのオプション

ディメンションテーブルのオプション

キャッシュの仕組み

ディメンションテーブルのキャッシュは、Tablestore への繰り返しルックアップを減らします。テーブルのサイズとクエリパターンに基づいてキャッシュポリシーを選択します。

• None：キャッシングなし。すべてのルックアップは Tablestore に直接アクセスします。データが頻繁に変更され、鮮度が重要な場合に使用します。

• LRU：最近アクセスされた一定数のレコードをキャッシュします。ルックアップしたデータがキャッシュに存在しない場合、コネクタは Tablestore にクエリを実行し、その結果でキャッシュを更新します。このポリシーを使用する場合は、cacheSize と cacheTTLMs を設定します。

• ALL (デフォルト)：ジョブが開始される前に、ディメンションテーブル全体をキャッシュにロードします。以降のすべてのルックアップはキャッシュからデータを取得します。キャッシュの有効期限が切れると (cacheTTLMs)、コネクタはすべてのデータをリロードします。テーブルが小さく、キーが見つからないルックアップが多いと予想される場合に ALL を使用します。ALL を使用する場合、結合ノードのメモリを増やしてください。キャッシュには、リモートテーブルの約 2 倍のサイズが必要です。

Tablestore からの読み取りと Tablestore への書き込み

この例では、Tunnel Service を使用して Tablestore ソーステーブルから注文データを読み取り、Tablestore シンクテーブルに書き込みます。シンクテーブルは、自動インクリメントプライマリキー列を使用します。

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;

ワイドカラムテーブルから時系列テーブルへの同期

この例では、ワイドカラムソーステーブルから読み取り、時系列シンクテーブルに書き込みます。シンクテーブルの tags 列は MAP<STRING, STRING> を使用してタグのキーと値のペアを保持し、storageType を 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'
);

-- 個々のタグ列からタグマップを構築し、時系列シンクテーブルに挿入します
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;