Realtime Compute for Apache Flink:ディメンションテーブルの JOIN ステートメント

チューニングパラメータのカテゴリと渡し方

ディメンションテーブル JOIN のチューニングには、3 つのカテゴリのパラメータが関係します。各カテゴリに固有の渡し方があり、渡し方を間違えると、ヒントは警告なしに無視されます。

OPTIONS ヒントによるコネクタオプションの上書き

OPTIONS() ヒントを使用すると、CREATE TABLE ステートメントを変更することなく、クエリ内で直接テーブルの WITH オプションを上書きできます。この方法は一般的にキャッシュの動作をチューニングするために使用されます。

SELECT t.id, t.name, w.phoneNumber
FROM kafka_input AS t
LEFT JOIN phoneNumber /*+ OPTIONS(
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '1000'
) */ FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

OPTIONS 内のすべてのキーは、コネクタが実際にサポートする WITH オプションでなければなりません。

LOOKUP ヒントによるルックアップ動作の設定

LOOKUP ヒントは、Apache Flink コミュニティ版と一貫した動作をします。これは、単一の結合操作の同期/非同期モード、リトライ、シャッフルの動作を設定します。詳細については、「Apache Flink Lookup Hint」をご参照ください。

• LOOKUP ヒントは、VVR 8.0 以降で利用可能です。

• 'shuffle' = 'true' オプションは、VVR 8.0.8 以降で利用可能です。

• エイリアスは VVR 8.0 以降でサポートされています。ディメンションテーブルがエイリアスで参照される場合、ヒントはそのエイリアスを使用する必要があります。

-- shuffle 戦略をディメンションテーブル dim1 にのみ適用します。
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- shuffle 戦略を dim1 と dim2 の両方に適用します。
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true'), LOOKUP('table'='dim2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- dim1 がエイリアス D1 で参照される場合、ヒントはエイリアスを使用する必要があります。
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

-- エイリアスを使用して、shuffle 戦略を dim1 と dim2 の両方に適用します。
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true'), LOOKUP('table'='D2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

SET を使用したジョブレベルの TableConfig 設定

table.exec.* および類似の TableConfig キーは、ジョブレベルのグローバル設定です。これらは SET を使用して渡す必要があり、LOOKUP ヒントも OPTIONS ヒントも受け付けません。

ディメンションテーブル JOIN でよく使われる TableConfig キー：

-- ジョブの開始時にジョブレベルのデフォルトを設定します。
SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO sink_table
SELECT ...
FROM src_table
LEFT JOIN dim_table FOR SYSTEM_TIME AS OF PROCTIME() AS d
ON src_table.key = d.key;

結合戦略ヒントによるシャッフルの制御

結合戦略ヒントは、ディメンションテーブル結合のシャッフル動作を制御します。利用可能なヒントは SHUFFLE_HASH、REPLICATED_SHUFFLE_HASH、SKEW です。以下の表は、どの結合戦略がどのキャッシュ戦略に適しているかを示しています。

-- dim1 にのみ SHUFFLE_HASH を有効にします。
SELECT /*+ SHUFFLE_HASH(dim1) */ ...

-- dim1 と dim2 の両方に SHUFFLE_HASH を有効にします。
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */ ...

-- dim1 がエイリアス D1 で参照される場合、ヒントはエイリアスを使用する必要があります。
SELECT /*+ SHUFFLE_HASH(D1) */ ...

SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ ...

SELECT /*+ SKEW(src) */  ...

例 1：基本的なディメンションテーブル JOIN

最も基本的な例：Kafka ストリームを MySQL ディメンションテーブルでエンリッチします。チューニングヒントは使用されません。

CREATE TEMPORARY TABLE kafka_input (
  id   BIGINT,
  name VARCHAR,
  age  BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE phoneNumber (
  name        VARCHAR,
  phoneNumber BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

CREATE TEMPORARY TABLE result_infor (
  id          BIGINT,
  phoneNumber BIGINT,
  name        VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO result_infor
SELECT
  t.id,
  w.phoneNumber,
  t.name
FROM kafka_input AS t
JOIN phoneNumber FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

例 2：OPTIONS ヒントでキャッシュを有効にし、LOOKUP ヒントで非同期ルックアップを設定

チューニングシナリオ：ディメンションテーブルの QPS が不十分です。このシナリオでは、ディメンションテーブルのパーシャルキャッシュを有効にし、より大きなバッファキューで非同期ルックアップを有効にします。

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true', 'capacity' = '3072') */
  t1.member_id AS member_id,
  t2.tag AS tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info /*+ OPTIONS(
    'lookup.cache' = 'PARTIAL',
    'lookup.partial-cache.max-rows' = '1000'
  ) */
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• コネクタオプション (lookup.cache、lookup.partial-cache.max-rows) は、ディメンションテーブル参照の直後に配置された OPTIONS() ヒントを使用して渡されます。これにより、クエリ時に対応する WITH オプションが上書きされます。

• ルックアップの動作 (async、capacity) は、SELECT の後に配置された LOOKUP() ヒントを使用して設定されます。OPTIONS と LOOKUP は 2 つの異なるヒントタイプであり、別々に記述する必要があります。これらを組み合わせると、ヒントは警告なしに無視されます。

例 3：SET を使用したジョブレベルの非同期ルックアップバッファのチューニング

チューニングシナリオ：同じジョブ内の複数のディメンションテーブル JOIN で、より大きな非同期ルックアップバッファが必要な状況です。この変更をグローバルに適用し、LOOKUP ヒントを使用して非同期を有効にします。

SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true') */
  t1.member_id,
  t2.tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• table.exec.async-lookup.buffer-capacity および同様のパラメータは、ジョブレベルの TableConfig キーであり、SET でのみ設定できます。OPTIONS または LOOKUP 内に配置しても効果はありません。

• LOOKUP ヒントが capacity を指定しない場合、SET で設定したグローバル値が使用されます。両方が指定されている場合、対象の結合では LOOKUP ヒントが優先されます。