Realtime Compute for Apache Flink:Tablestore

Opsi umum

Semua tipe tabel berbagi opsi berikut.

Opsi tabel sumber

Opsi tabel sink

Opsi tabel dimensi

Cara kerja cache

Cache tabel dimensi mengurangi pencarian berulang terhadap Tablestore. Pilih kebijakan cache berdasarkan ukuran tabel dan pola kueri Anda:

• None: Tanpa caching. Setiap pencarian langsung mengakses Tablestore. Gunakan ketika data sering berubah dan freshness sangat penting.

• LRU: Menyimpan jumlah tetap catatan yang baru saja diakses. Ketika pencarian tidak ditemukan di cache, connector melakukan kueri ke Tablestore dan memperbarui cache dengan hasilnya. Atur cacheSize dan cacheTTLMs saat menggunakan kebijakan ini.

• ALL (default): Memuat seluruh tabel dimensi ke dalam cache sebelum pekerjaan dimulai. Semua pencarian selanjutnya dilayani dari cache. Saat cache kedaluwarsa (cacheTTLMs), connector memuat ulang semua data. Gunakan ALL ketika tabel kecil dan Anda mengharapkan banyak pencarian dengan kunci yang tidak ditemukan. Saat menggunakan ALL, tingkatkan memori node join — cache memerlukan ruang sekitar dua kali ukuran tabel remote.

Baca dari Tablestore dan tulis ke Tablestore

Contoh ini membaca data pesanan dari tabel sumber Tablestore melalui Tunnel Service dan menulisnya ke tabel sink Tablestore. Tabel sink menggunakan kolom kunci utama auto-increment.

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;

Menyinkronkan tabel wide-column ke tabel time series

Contoh ini membaca dari tabel sumber wide-column dan menulis ke tabel sink time series. Kolom tags pada tabel sink menggunakan MAP<STRING, STRING> untuk menyimpan pasangan kunci-nilai tag, dan storageType diatur ke 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'
);

-- Bangun map tags dari kolom tag individual dan masukkan ke tabel sink time series
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;