注意事項

在作業啟動前，如果未在運行參數配置區域顯式設置該參數，在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。