Realtime Compute for Apache Flink:ApsaraDB for HBase

Contoh tabel dimensi

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  `proc_time` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE hbase_dim (
  rowkey INT,
  family1 ROW<col1 INT>,
  family2 ROW<col1 STRING, col2 BIGINT>,
  family3 ROW<col1 DOUBLE, col2 BOOLEAN, col3 STRING>
) WITH (
  'connector' = 'cloudhbase',
  'table-name' = '<yourTableName>',
  'zookeeper.quorum' = '<yourZookeeperQuorum>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  f1c1 INT,
  f3c3 STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT a, family1.col1 AS f1c1, family3.col3 AS f3c3
FROM datagen_source
JOIN hbase_dim FOR SYSTEM_TIME AS OF datagen_source.`proc_time` AS h
ON datagen_source.a = h.rowkey;

Contoh tabel sink

CREATE TEMPORARY TABLE datagen_source (
  rowkey INT,
  f1q1 INT,
  f2q1 STRING,
  f2q2 BIGINT,
  f3q1 DOUBLE,
  f3q2 BOOLEAN,
  f3q3 STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE hbase_sink (
  rowkey INT,
  family1 ROW<q1 INT>,
  family2 ROW<q1 STRING, q2 BIGINT>,
  family3 ROW<q1 DOUBLE, q2 BOOLEAN, q3 STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'cloudhbase',
  'table-name' = '<yourTableName>',
  'zookeeper.quorum' = '<yourZookeeperQuorum>'
);

INSERT INTO hbase_sink
SELECT rowkey, ROW(f1q1), ROW(f2q1, f2q2), ROW(f3q1, f3q2, f3q3)
FROM datagen_source;

Sink kolom dinamis

Ketika dynamic.table diatur ke true, setiap keluarga kolom harus mendeklarasikan tepat dua bidang: yang pertama merepresentasikan nama kolom dinamis, dan yang kedua merepresentasikan nilainya.

CREATE TEMPORARY TABLE datagen_source (
  id INT,
  f1hour STRING,
  f1deal BIGINT,
  f2day STRING,
  f2deal BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE hbase_sink (
  rowkey INT,
  f1 ROW<`hour` STRING, deal BIGINT>,
  f2 ROW<`day` STRING, deal BIGINT>
) WITH (
  'connector' = 'cloudhbase',
  'table-name' = '<yourTableName>',
  'zookeeper.quorum' = '<yourZookeeperQuorum>',
  'dynamic.table' = 'true'
);

INSERT INTO hbase_sink
SELECT id, ROW(f1hour, f1deal), ROW(f2day, f2deal)
FROM datagen_source;

Sebagai contoh, jika baris sumber memiliki id=1, f1hour='10', f1deal=100, f2day='2020-7-26', dan f2deal=10000, maka baris HBase hasilnya memiliki rowkey=1, f1:10=100, dan f2:2020-7-26=10000.

Opsi umum

Opsi sink

Opsi tabel dimensi

Kebijakan cache:

• None — Tanpa caching. Setiap lookup langsung mengakses ApsaraDB for HBase.

• LRU — Mencache subset baris. Saat terjadi cache miss, sistem langsung mengakses ApsaraDB for HBase. Konfigurasikan cacheSize dan cacheTTLMs saat menggunakan kebijakan ini.

• ALL (default) — Memuat seluruh tabel dimensi ke dalam cache sebelum pekerjaan dimulai. Semua lookup selanjutnya mengenai cache. Jika kunci tidak ditemukan di cache, artinya kunci tersebut tidak ada di tabel sumber. Cache dimuat ulang setelah kedaluwarsa. Konfigurasikan cacheTTLMs dan opsional cacheReloadTimeBlackList saat menggunakan kebijakan ini.

  Ketika cache diatur ke ALL, node yang melakukan join memerlukan memori tambahan — kira-kira dua kali ukuran tabel dimensi — karena sistem memuat semua data secara asinkron sebelum pekerjaan dimulai. Hal ini dapat memperlambat startup pekerjaan.