Realtime Compute for Apache Flink:主要パラメーター

注意事項

ジョブの開始前に、[パラメーター] セクションでこのパラメーターを明示的に設定しない場合、ミニバッチモードではマネージドメモリを使用してデータをバッファリングします。次のいずれかの条件で最終計算とデータ出力がトリガーされます。

• MiniBatchAssigner オペレーターからウォーターマークメッセージを受信した場合。

• マネージドメモリがフルになった場合。

• チェックポイントの開始前。

• ジョブが停止した場合。

使用可能な値

• -1 (デフォルト)：マネージドメモリを使用してデータをバッファリングすることを示します。

• その他の負の Long 値：デフォルト設定と同じです。

• その他の正の Long 値：ヒープメモリを使用してデータをバッファリングすることを示します。バッファリングされたレコード数がこの値 (N) に達すると、システムは出力操作を自動的にトリガーします。

例

1. SQL ストリーミングジョブを作成し、次の SQL コードをコピーして、ジョブを [Deploy] します。

   CREATE TEMPORARY TABLE s1 (
     a INT,
     b INT,
     ts TIMESTAMP(3),
     PRIMARY KEY (a) NOT ENFORCED,
     WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
   ) WITH (
     'connector'='datagen',
     'rows-per-second'='1',
     'fields.ts.kind'='random',
     'fields.ts.max-past'='5s',
     'fields.b.kind'='random',
     'fields.b.min'='0',
     'fields.b.max'='10'
   );
   CREATE TEMPORARY TABLE sink (
     a INT,
     b BIGINT,
     PRIMARY KEY (a) NOT ENFORCED
   ) WITH (
     'connector'='print'
   );
   INSERT INTO sink SELECT a, sum(b) FROM s1 GROUP BY a;

2. [ジョブ O&M] ページの [デプロイ詳細] タブで、[ランタイムパラメーター設定] セクションの [その他の設定] フィールドに、table.exec.mini-batch.enabled: true と table.exec.mini-batch.allow-latency: 2s パラメーターを設定し、table.exec.mini-batch.size はデフォルト値の -1 を使用するため未設定のままにします。

3. ジョブを [開始] します。 [ステータス] タブでは、ジョブトポロジに MiniBatchAssigner、LocalGroupAggregate、および GlobalGroupAggregate オペレーターが含まれます。

注意事項

• このパラメーターは VVR 8.0.8 以降でのみ有効です。VVR 8.0.8 より前のバージョンでは、このパラメーターを false に設定した場合と同等の動作になります。

• 値を false から true に変更すると、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターから下流に送信されるデータ量が増加し、下流オペレーターの負荷が高まる可能性があります。

使用可能な値

• false (デフォルト)：ステート TTL が有効な場合、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターは、集計結果が変化しないときに下流にデータを出力しません。

• true：ステート TTL が有効な場合、集計結果が変化しなくても、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターは更新 (重複) レコードを下流に引き続き出力します。

例

1. SQL ストリーミングジョブを作成し、次の SQL コードをコピーして、ジョブを [Deploy] します。

   create temporary table src(
       a int,
       b string
   ) with (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.a.min' = '1',
       'fields.a.max' = '1',
       'fields.b.length' = '3'
   );
   create temporary table snk(
       a int,
       max_length_b bigint
   ) with (
       'connector' = 'blackhole'
   );
   insert into snk select a, max(CHAR_LENGTH(b)) from src group by a; 

2. [ジョブ O&M] ページの [デプロイ詳細] タブの [ランタイムパラメーター設定] エリアの [その他の設定] セクションで、table.exec.mini-batch.enabled: true と table.exec.mini-batch.allow-latency: 2s パラメーターを設定して、Minibatch Aggregate の最適化を有効にします。

3. ジョブを[開始]します。[ステータス] タブでは、ジョブに MinibatchGlobalAggregate 演算子が含まれています。演算子の「+」記号をクリックすると、集計結果が変更されない場合に GlobalGroupAggregate 演算子がデータをダウンストリームに送信しないことを確認できます。

   オペレーターの RecordsIn が 19、RecordsOut が 1 と表示されます。これは、19 件の入力レコードから 1 件の集計出力のみが生成されたことを意味します。

4. ジョブを停止し、[ジョブ O&M] ページの [デプロイ詳細] ページにある [実行パラメータ設定] セクションの [その他の設定] に、パラメーター table.exec.agg.mini-batch.output-identical-enabled: true を追加します。

5. ジョブを[開始] します。[ステータス] タブでは、ジョブに MinibatchGlobalAggregate オペレーターが含まれていることを確認できます。オペレーターの「+」記号をクリックすると、集計結果が変更されていない場合でも GlobalGroupAggregate オペレーターがダウンストリームにデータを送信することを確認できます。ジョブを再起動すると、[ステータス] タブに、[GlobalGroupAggregate] オペレーターの [RecordsIn] と [RecordsOut] が両方とも 94 であることが表示されます。これは、table.exec.agg.mini-batch.output-identical-enabled: true が有効になっている場合、集計結果が変更されなくてもオペレーターがダウンストリームにデータを送信することを示しています。

ユースケース

• 同じ更新キーを持つメッセージが時間とともに少なくなるストリームにおいて、ディメンションテーブル結合でキーごとの処理順序を保持するためにこの最適化を使用します。

• プライマリキーを持つ変更データキャプチャ (CDC) ストリームで、ディメンションテーブル結合によるデータエンリッチメントを行い、シンクのプライマリキーがソースのプライマリキーと一致するシンクに書き込みます。ディメンションテーブル結合の結合キーはプライマリキーとは異なり、ディメンション側の結合キーがプライマリキーです。この最適化では、更新キーとして派生した CDC プライマリキーによってシャッフルが行われます。同じシナリオで SHUFFLE_HASH 最適化を有効にする場合と比較して、この方法では、より高い並列度でシンクの前に SinkMaterializer オペレーターが生成されるのを回避します。これにより、そのオペレーターに起因する潜在的なパフォーマンスの問題、特に長時間実行中に蓄積される可能性のある大きなステートを回避できます。SinkUpsertMaterializer の詳細については、「推奨事項」をご参照ください。

• ディメンションテーブル結合の結合キーがプライマリキーと異なり、ディメンション側の結合キーがプライマリキーであり、その後にランク演算子が続く場合、この最適化は、更新キーとして導出される CDC プライマリキーによってシャッフルを行うことで、同じシナリオで SHUFFLE_HASH 最適化を有効にする場合と比較して UpdateFastRank が RetractRank に低下するのを防ぎます。RetractRank を UpdateFastRank に最適化する方法の詳細については、「TopN 最適化手法」をご参照ください。

注意事項

• ストリームに更新キーがない場合、行全体がキーとして使用されます。

• 同じ更新キーが短時間に頻繁に更新される場合、同じ更新キーのレコードが厳密な順序で処理されるため、スループットが低下します。

• 元の非同期ディメンションテーブル結合と比較すると、Key-Ordered モードではキー付きステートが導入されます。このモードの有効化/無効化はステート互換性に影響します。

• この機能は、ディメンションテーブル結合の入力が更新ストリームであり、かつ table.exec.async-lookup.output-mode='ALLOW_UNORDERED' と table.exec.async-lookup.key-ordered-enabled='true' を設定した場合にのみ、VVR 8.0.10 以降で有効になります。

使用可能な値

• false (デフォルト)：Key-Ordered モードを無効にします。

• true：Key-Ordered モードを有効にします。

例

1. 次の例では、非同期 Hologres ディメンションテーブル結合を使用します。SQL ストリーミングジョブを作成し、次の SQL コードをコピーして、ジョブを [Deploy] します。

   Hologres コネクタの詳細については、「Hologres」をご参照ください。

   CREATE TEMPORARY TABLE bid_source(
     auction  BIGINT,
     bidder  BIGINT,
     price  BIGINT,
     channel  VARCHAR,
     url  VARCHAR,
     dateTime  TIMESTAMP(3),
     extra  VARCHAR,
     proc_time AS proctime(),
     WATERMARK FOR dateTime AS dateTime - INTERVAL '4' SECOND
   ) WITH (
     'connector' = 'kafka',  -- insert-only ではないストリームコネクター
     'topic' = 'user_behavior',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'canal-json'
   );
   CREATE TEMPORARY TABLE users (
       user_id STRING PRIMARY KEY NOT ENFORCED, -- プライマリキーを定義
       user_name VARCHAR(255) NOT NULL, 
       age INT NOT NULL
   ) WITH (
       'connector' = 'hologres',    -- 非同期ルックアップをサポートするコネクター
       'async' = 'true',
       'dbname' = 'holo db name', -- Hologres データベースの名前
       'tablename' = 'schema_name.table_name', -- データを受信する Hologres テーブルの名前
       'username' = 'access id', -- Alibaba Cloud アカウントの AccessKey ID
       'password' = 'access key', -- Alibaba Cloud アカウントの AccessKey Secret
       'endpoint' = 'holo vpc endpoint' -- Hologres インスタンスの VPC エンドポイント
   );
   CREATE TEMPORARY TABLE bh ( 
       auction  BIGINT,
       age INT
   ) WITH (
       'connector' = 'blackhole'
   );
   INSERT INTO bh
   SELECT
       bid_source.auction,
       u.age
   FROM bid_source
       JOIN users FOR SYSTEM_TIME AS OF bid_source.proc_time AS u
       ON bid_source.channel = u.user_id;
   

2. [ジョブ O&M] ページの [デプロイ詳細] タブ、[ランタイムパラメーター設定] エリアの [その他の設定] セクションで、table.exec.async-lookup.output-mode='ALLOW_UNORDERED' および table.exec.async-lookup.key-ordered-enabled='true' パラメーターを設定します。

3. ジョブを [開始] します。 [ステータス] タブでは、ジョブの async 属性が KEY_ORDERED:true であることを確認できます。

table.optimizer.window-join-enabled

このパラメーターは、ウィンドウ結合操作を有効にするかどうかを制御します。有効にすると、Flink は対応する実行計画を ウィンドウ結合として最適化します。小さなウィンドウでは、これによりステートのオーバーヘッドが削減され、パフォーマンスが向上します。通常結合と比較して、ウィンドウ結合では下流オペレーターに更新メッセージを出力しないことも可能であり、時間ウィンドウが小さい結合が必要なユースケースで有用です。