Key parameters

Usage notes

Before a job starts, if you do not explicitly set this parameter in the Parameters section, managed memory is used to buffer data in mini-batch mode. Any of the following conditions triggers the final computation and data output:

• A watermark message is received from the MiniBatchAssigner operator.

• The managed memory is full.

• Before a checkpoint starts.

• The job is stopped.

Available values

• -1 (default): Indicates that managed memory is used to buffer data.

• Other negative Long values: Same as the default setting.

• Other positive Long values: Indicates that heap memory is used to buffer data. When the number of buffered records reaches this value (N), the system automatically triggers the output operation.

Example

1. Create an SQL streaming job, copy the following SQL code, and deploy the job.

   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. On the Deployment Details tab of the Job O&M page, in the Other Configurations field of the Runtime Parameter Configuration section, set the table.exec.mini-batch.enabled: true and table.exec.mini-batch.allow-latency: 2s parameters, and do not set table.exec.mini-batch.size to use its default value of -1.

3. Start the job. On the Status tab, the job topology includes the MiniBatchAssigner, LocalGroupAggregate, and GlobalGroupAggregate operators.

Usage notes

• This parameter is effective only in VVR 8.0.8 and later. In versions before VVR 8.0.8, the behavior is equivalent to setting this parameter to false.

• When changing the value from false to true, the amount of data sent downstream from the MinibatchGlobalAgg and MinibatchAgg operators may increase, putting more pressure on downstream operators.

Available values

• false (default): When state TTL is enabled, the MinibatchGlobalAgg and MinibatchAgg operators do not emit data downstream if the aggregation result does not change.

• true: When state TTL is enabled, the MinibatchGlobalAgg and MinibatchAgg operators still emit updated (duplicate) records downstream even if the aggregation result does not change.

Example

1. Create an SQL streaming job, copy the following SQL code, and deploy the job.

   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. In the Other Configurations section of the Runtime Parameter Configuration area on the Deployment Details tab of the Job O&M page, set the table.exec.mini-batch.enabled: true and table.exec.mini-batch.allow-latency: 2s parameters to enable Minibatch Aggregate optimization.

3. Start the job. On the Status tab, the job includes a MinibatchGlobalAggregate operator. Click the "+" sign on the operator to see that the GlobalGroupAggregate operator does not send data downstream when the aggregation result is unchanged.

   The operator shows RecordsIn as 19 and RecordsOut as 1, which means that 19 input records produced only 1 aggregate output.

4. Stop the job, and add the parameter table.exec.agg.mini-batch.output-identical-enabled: true to Other Configurations in the Running Parameter Configuration section on the Deployment Details page of the Job O&M page.

5. Start the job. On the Status tab, you can see the job includes a MinibatchGlobalAggregate operator. Click the "+" sign on the operator to observe that the GlobalGroupAggregate operator now sends data downstream even when the aggregation result is unchanged. After restarting the job, the Status tab shows that the GlobalGroupAggregate operator's RecordsIn and RecordsOut are both 94. This indicates that with table.exec.agg.mini-batch.output-identical-enabled: true enabled, the operator sends data downstream even if the aggregation result does not change.

Use cases

• Use this optimization to preserve per-key processing order in a dimension table join when the stream has few messages with the same update key over time.

• In a Change Data Capture (CDC) stream with a primary key, you perform data enrichment with a dimension table join and write to a sink where the sink primary key matches the source primary key. The join key for the dimension table join is different from the primary key, and the dimension-side join key is the primary key. This optimization shuffles by the CDC primary key, which is derived as the update key. Compared with enabling the SHUFFLE_HASH optimization for the same scenario, it avoids generating a SinkMaterializer operator before the sink at higher parallelism. This avoids potential performance issues from that operator, especially the large state that can accumulate during long runs. For information about SinkUpsertMaterializer, see Recommendations.

• The join key for the dimension table join is different from the primary key, the dimension-side join key is the primary key, and a rank operator follows. This optimization shuffles by the CDC primary key, which is derived as the update key. Compared with enabling the SHUFFLE_HASH optimization for the same scenario, it prevents UpdateFastRank from degrading to RetractRank. For information about how to optimize RetractRank to UpdateFastRank, see TopN optimization techniques.

Usage notes

• If the stream has no update key, the entire row is used as the key.

• Throughput decreases if the same update key is updated frequently within a short period, because records for the same update key are processed in strict order.

• Compared with the original asynchronous dimension table join, Key-Ordered mode introduces keyed state. Enabling or disabling this mode affects state compatibility.

• This feature takes effect only for VVR 8.0.10 and later versions when the input of the dimension table join is an update stream and you configure table.exec.async-lookup.output-mode='ALLOW_UNORDERED' and table.exec.async-lookup.key-ordered-enabled='true'.

Available values

• false (default): Disables Key-Ordered mode.

• true: Enables Key-Ordered mode.

Example

1. The following example uses an asynchronous Hologres dimension table join. Create an SQL streaming job, copy the following SQL code, and deploy the job.

   For more information about the Hologres connector, see 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',  -- A non-insert-only stream connector
     '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, -- Define the primary key
       user_name VARCHAR(255) NOT NULL, 
       age INT NOT NULL
   ) WITH (
       'connector' = 'hologres',    -- A connector that supports asynchronous lookup
       'async' = 'true',
       'dbname' = 'holo db name', --The name of your Hologres database
       'tablename' = 'schema_name.table_name', --The name of the Hologres table to receive data
       'username' = 'access id', --The AccessKey ID of your Alibaba Cloud account
       'password' = 'access key', --The AccessKey Secret of your Alibaba Cloud account
       'endpoint' = 'holo vpc endpoint', --The VPC endpoint of your Hologres instance
   );
   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. On the Job O&M page, on the Deployment Details tab, in the Other Configurations section of the Runtime Parameter Configuration area, set the table.exec.async-lookup.output-mode='ALLOW_UNORDERED' and table.exec.async-lookup.key-ordered-enabled='true' parameters.

3. Start the job. On the Status tab, you can see that the job's async attribute is KEY_ORDERED:true.

table.optimizer.window-join-enabled

This parameter controls whether to enable window join operations. When enabled, Flink optimizes the corresponding execution plan as a window join. For small windows, this reduces state overhead and improves performance. Compared with a regular join, a window join can also avoid emitting update messages to downstream operators, which is useful for use cases that require joins with small time-window conditions.