Realtime Compute for Apache Flink：重要參數說明

注意事項

在作業啟動前，如果未在運行參數配置地區顯式設定該參數，在MiniBatch處理模式下，將使用Managed Memory快取資料，在以下幾種條件下都會觸發最終計算和資料輸出：

• 收到MiniBatchAssigner節點發送的watermark訊息

• Managed Memory已滿

• 進行Checkpoint前

• 作業停止時

取值說明

• -1（預設值）：表示使用Managed Memory快取資料。

• 其他Long類型的負值：同預設設定。

• 其他Long類型的正值：表示使用Heap Memory來快取資料。當緩衝的資料量達到N條時，會自動觸發輸出操作。

使用樣本

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. 在作業營運頁面的部署詳情頁運行參數配置地區其他配置中，設定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（預設值）：表示在開啟State TTL的情況下，MinibatchGlobalAgg節點和MinibatchAgg節點在消費資料後，如果彙總結果未發生變化，就不向下遊下發資料。

• true：表示在開啟State 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. 在作業營運頁面的部署詳情頁運行參數配置地區其他配置中，設定table.exec.mini-batch.enabled: true和table.exec.mini-batch.allow-latency: 2s參數，啟用Minibatch Aggregate最佳化。

3. 啟動作業。在狀態總覽頁簽下，您可以看到作業包含了MinibatchGlobalAggregate節點，點擊該節點上的“+”號，可以觀察到GlobalGroupAggregate節點在彙總結果不變的情況下，不向下遊發送資料。

   

4. 停止該作業，在作業營運頁面的部署詳情頁運行參數配置地區其他配置中，添加參數table.exec.agg.mini-batch.output-identical-enabled: true

5. 啟動作業。在狀態總覽頁簽下，您可以看到作業包含了MinibatchGlobalAggregate節點，點擊該節點上的“+”號，可以觀察到GlobalGroupAggregate節點在彙總結果不變的情況下，仍然向下遊發送資料。

應用情境

• 一段時間內流上相同更新鍵的訊息量較少（比如更新鍵為主鍵的情境，相同主鍵的資料更新頻率不高），同時在維表join時對基於更新鍵的處理有處理順序的需求。該最佳化可以保證基於更新鍵的資料處理順序。

• 在包含主鍵的CDC流中，通過維表join打寬寫入Sink（Sink的主鍵與Source的主鍵保持一致），且維表join的join key和主鍵不一致，維表側join key為主鍵。該最佳化會根據CDC主鍵（被推導為更新鍵）進行shuffle。相比同情境開啟SHUFFLE_HASH最佳化，在多並發的情況下，可以避免在Sink前產生SinkMaterializer 節點，從而消除因該節點引起的潛在效能問題，尤其可以消除長期運行時該節點產生的大state。有關SinkUpsertMaterializer請參見使用建議。

• 維表join的join key和主鍵不一致，維表側join key為主鍵，且之後存在rank節點，該最佳化會根據CDC主鍵（被推導為更新鍵）進行shuffle。相比同情境開啟SHUFFLE_HASH最佳化，可以避免UpdateFastRank退化為RetractRank。RetractRank如何能最佳化成UpdateFastRank請參見TopN最佳化技巧。

注意事項

• 當流上不存在更新鍵時，會將整行資料作為key。

• 在短時間內同一個更新鍵存在較頻繁的更新時輸送量會降低，因為針對同一個更新鍵的資料是嚴格按照順序處理的。

• Key-Ordered模式相比原有非同步維表Join新增了Keyed State，開啟或關閉該模式會影響狀態相容性。

• 僅適用於VVR 8.0.10及以上版本且維表Join的輸入是非更新流，配置table.exec.async-lookup.output-mode='ALLOW_UNORDERED'、table.exec.async-lookup.key-ordered-enabled='true'時才會生效。

取值說明

• false（預設值）：表示不開啟Key-Ordered模式。

• true：表示開啟Key-Ordered模式。

使用樣本

1. 以使用Hologres非同步維表Join為例，建立SQL流作業，複製如下測試SQL，部署作業。

   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' = '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',    -- 支援非同步lookup功能連機器
       'async' = 'true',
       'dbname' = 'holo db name', --Hologres的資料庫名稱
       'tablename' = 'schema_name.table_name', --Hologres用於接收資料的表名稱
       'username' = 'access id', --當前阿里雲帳號的AccessKey ID
       'password' = 'access key', --當前阿里雲帳號的AccessKey Secret
       'endpoint' = 'holo vpc endpoint', --當前Hologres執行個體VPC網路的Endpoint
   );
   
   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。