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

使用上の注意

ジョブが開始される前に [パラメーター] セクションでこのパラメーターを明示的に構成しない場合、miniBatch 処理モードでは管理メモリを使用してデータがキャッシュされます。次のいずれかの条件が満たされると、最終的な計算とデータ出力がトリガーされます。

• 計算オペレーターが MiniBatchAssigner オペレーターから送信されたウォーターマークメッセージを受信する。

• 管理メモリがいっぱいになる。

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

• ジョブがキャンセルされる。

有効な値

• -1 (デフォルト): 管理メモリを使用してデータがキャッシュされます。

• LONG 型の負の値：処理メカニズムは、デフォルト値の処理メカニズムと同じです。

• LONG 型の正の値：ヒープメモリを使用してデータがキャッシュされます。キャッシュされた入力レコード数が指定された値に達すると、データ出力がトリガーされます。

例

1. SQL ストリーミングドラフトを作成し、次のテスト SQL 文をコピーして、ドラフトをデプロイします。

   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. MiniBatch サイズを構成します。

   1. ジョブ詳細ページに移動します。

   2. [構成] タブで、[パラメーター] セクションを見つけ、[編集] をクリックします。

   3. [その他の構成] で、table.exec.mini-batch.enabled: true および table.exec.mini-batch.allow-latency: 2s を設定します。

   4. table.exec.mini-batch.size を -1 に設定するか、このパラメーターをスキップします。

3. ジョブを開始します。

   [ステータス] タブで、ジョブのトポロジーには MiniBatchAssigner、LocalGroupAggregate、および GlobalGroupAggregate オペレーターが含まれています。

   

使用上の注意

• このスイッチは、Ververica Runtime (VVR) 8.0.8 以降でのみ有効です。

• 値を false から true に変更すると、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターによって送信されるデータ量が増加する可能性があります。これにより、ダウンストリームオペレーターに負荷がかかる可能性があります。

有効な値

• false (デフォルト): 状態データの TTL が有効で、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターによってデータが消費された後も集約結果が変更されない場合、重複データはダウンストリームオペレーターに送信されません。

• true: 状態データの TTL が有効で、MinibatchGlobalAgg オペレーターと MinibatchAgg オペレーターによってデータが消費された後も集約結果が変更されない場合、重複データはダウンストリームオペレーターに送信されます。

例

1. SQL ストリーミングドラフトを作成し、次のテスト SQL 文をコピーして、ドラフトをデプロイします。

   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. MiniBatch 集約最適化を有効にします。

   1. ジョブ詳細ページに移動します。

   2. [構成] タブで、[パラメーター] セクションを見つけ、[編集] をクリックします。

   3. [その他の構成] に、table.exec.mini-batch.enabled: true および table.exec.mini-batch.allow-latency: 2s を追加します。

3. ジョブを開始します。

   [ステータス] タブに、MinibatchGlobalAggregate オペレーターが表示されます。オペレーターの + をクリックして、集約結果が変更されない場合、GlobalGroupAggregate オペレーターがダウンストリームオペレーターにデータを送信していないことを確認します。

   

4. ジョブを停止します。[構成] タブで、[パラメーター] セクションを見つけ、[編集] をクリックします。[その他の構成] に、table.exec.agg.mini-batch.output-identical-enabled: true を追加します。

5. ジョブを開始します。

   [ステータス] タブに、MinibatchGlobalAggregate オペレーターが表示されます。オペレーターの + をクリックして、集約結果が変更されない場合、GlobalGroupAggregate オペレーターがダウンストリームオペレーターにデータを送信することを確認します。

シナリオ

• 一定期間にわたってストリーム内の同じ更新キーを持つメッセージの数が少ない。たとえば、更新キーはプライマリキーと見なされ、同じプライマリキーを持つデータは頻繁に更新されません。さらに、ディメンションテーブルが結合される場合、更新キーに基づく処理順序が必要です。この場合、table.exec.async-lookup.key-ordered-enabled パラメーターを指定することで最適化を実行できます。これにより、更新キーに基づいたデータ処理順序を確保できます。

• プライマリキーを含む Change Data Capture (CDC) ストリームでは、ディメンションテーブルが結合されて、シンクへのデータ書き込み用のワイドテーブルが生成されます。シンクのプライマリキーは、ソースのプライマリキーと一致しています。さらに、ディメンションテーブルの JOIN 操作の結合キーは、プライマリキーと一致しません。結合キーはプライマリキーと見なされます。この場合、table.exec.async-lookup.key-ordered-enabled パラメーターを指定することで最適化を実行できます。これにより、システムは更新キーと見なされる CDC プライマリキーに基づいてデータをシャッフルできます。また、SHUFFLE_HASH 結合ポリシーを有効にして、このシナリオを最適化することもできます。高並列シナリオでは、この方法と比較して、table.exec.async-lookup.key-ordered-enabled パラメーターを指定することで最適化を実行すると、データがシンクに書き込まれる前に SinkMaterializer オペレーターが生成されるのを防ぐことができます。これにより、オペレーターによって引き起こされる可能性のあるパフォーマンスの問題、特にオペレーターの長時間実行中に大量の状態データが生成される問題を防ぐことができます。SinkUpsertMaterializer オペレーターの詳細については、「使用上の注意」をご参照ください。

• ディメンションテーブルの JOIN 操作の結合キーは、プライマリキーと一致しません。ディメンションテーブルの結合キーはプライマリキーと見なされ、JOIN 操作の実行後にランクオペレーターを使用できます。この場合、table.exec.async-lookup.key-ordered-enabled パラメーターを指定することで最適化を実行すると、システムは更新キーと見なされる CDC プライマリキーに基づいてデータをシャッフルできます。また、SHUFFLE_HASH 結合ポリシーを有効にして、このシナリオを最適化することもできます。この方法と比較して、table.exec.async-lookup.key-ordered-enabled パラメーターを指定することで最適化を実行すると、UpdateFastRank が RetractRank にダウングレードするのを防ぐことができます。RetractRank を UpdateFastRank に変更する方法の詳細については、「TopN の実践」をご参照ください。

使用上の注意

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

• 短期間に同じ更新キーが頻繁に更新されると、スループットが低下します。これは、同じ更新キーを持つデータが厳密な順序で処理されるためです。

• 最適化前の非同期モードでのディメンションテーブルの JOIN 操作と比較して、キー順序付きモードはキー付き状態を提供します。キー順序付きモードを有効または無効にすると、状態データの互換性に影響します。

• この最適化は、VVR 8.0.10 以降で table.exec.async-lookup.output-mode='ALLOW_UNORDERED' および table.exec.async-lookup.key-ordered-enabled='true' 構成をディメンションテーブルの JOIN 操作に追加し、入力ストリームが更新ストリームである場合にのみ有効になります。

有効な値

• false (デフォルト): キー順序モードを無効にします。

• true: キー順序モードを有効にします。

例

1. 次のコードをストリーミングドラフトにコピーして貼り付け、デプロイします。

   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' = 'csv'
   );
   
   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 シークレット。
       '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. ジョブ詳細ページに移動します。[構成] タブで、[パラメーター] セクションを見つけ、[編集] をクリックします。[その他の構成] に、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 はウィンドウ結合アプローチに基づいて実行計画を最適化します。小さなウィンドウの場合、これにより状態サイズを削減し、パフォーマンスを向上させることができます。さらに、通常の結合 とは異なり、ウィンドウ結合を有効にすると、更新メッセージがダウンストリームオペレーターに送信されるのを回避できます。これは、小さなウィンドウでの結合が必要なユースケースに適しています。