Realtime Compute for Apache Flink:ApsaraDB for HBase

ディメンションテーブルの例

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;

シンクテーブルの例

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

dynamic.table が true に設定されている場合、各カラムファミリーは正確に 2 つのフィールドを宣言する必要があります。最初のフィールドは動的カラム名を表し、2 番目のフィールドはその値を表します。

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;

たとえば、ソース行に id=1、f1hour='10'、f1deal=100、f2day='2020-7-26'、および f2deal=10000 がある場合、結果の HBase 行は rowkey=1、f1:10=100、および f2:2020-7-26=10000 となります。

一般オプション

sink オプション

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

キャッシュポリシー：

• None — キャッシュなし。すべてのルックアップは ApsaraDB for HBase に直接クエリを実行します。

• LRU — 行のサブセットをキャッシュします。キャッシュミスが発生した場合、システムは ApsaraDB for HBase に直接クエリを実行します。このポリシーを使用する場合は、cacheSize と cacheTTLMs を設定してください。

• ALL (デフォルト) — ジョブが開始される前に、ディメンションテーブル全体をキャッシュにロードします。以降のすべてのルックアップはキャッシュにヒットします。キーがキャッシュに見つからない場合、そのキーはソーステーブルに存在しません。キャッシュは有効期限が切れると再読み込みされます。このポリシーを使用する場合は、cacheTTLMs と、オプションで cacheReloadTimeBlackList を設定してください。

  cache が ALL に設定されている場合、ジョブが開始される前にシステムがすべてのデータを非同期でロードするため、結合を実行するノードには追加のメモリ (ディメンションテーブルのサイズの約 2 倍) が必要になります。これにより、ジョブの起動が遅くなる可能性があります。