Realtime Compute for Apache Flink:Key parameters

Usage notes

If you do not explicitly configure this parameter in the Parameters section before a job is started, managed memory is used to cache data in miniBatch processing mode. If one of the following conditions is met, final calculation and data output are triggered:

• The compute operator receives the watermark message sent by the MiniBatchAssigner operator.

• Managed memory is full.

• Before checkpointing.

• The job is canceled.

Valid values

• -1 (default): The managed memory is used to cache data.

• A negative value of the LONG type: The processing mechanism is the same as the processing mechanism for the default value.

• A positive value of the LONG type: The heap memory is used to cache data. When the number of cached input records reaches the specified value, data output is triggered.

Example

1. Create an SQL streaming draft, copy the following test SQL statements, and then deploy the draft.

   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. Configure the MiniBatch size.

   1. Go to the job details page.

   2. On the Configuration tab, find the Parameters section, and click Edit.

   3. In Other Configuration, set table.exec.mini-batch.enabled: true and table.exec.mini-batch.allow-latency: 2s.

   4. Set table.exec.mini-batch.size to -1 or skip this parameter.

3. Start the job.

   On the Status tab, the topology of the job contains the MiniBatchAssigner, LocalGroupAggregate, and GlobalGroupAggregate operators.

   

Usage notes

• This switch takes effect only in Ververica Runtime (VVR) 8.0.8 or later.

• If you change the value from false to true, the amount of data sent by the MinibatchGlobalAgg and MinibatchAgg operators may increase. This may cause pressure on downstream operators.

Valid values

• false (default): If the TTL of state data is enabled and the aggregation result remains unchanged after data is consumed by the MinibatchGlobalAgg and MinibatchAgg operators, duplicate data is not sent to downstream operators.

• true: If the TTL of state data is enabled and the aggregation result remains unchanged after data is consumed by the MinibatchGlobalAgg and MinibatchAgg operators, duplicate data is sent to downstream operators.

Example

1. Create an SQL streaming draft, copy the following test SQL statements, and then deploy the draft.

   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. Enable MiniBatch aggregate optimization.

   1. Go to the job details page.

   2. On the Configuration tab, find the Parameters section and click Edit.

   3. In Other Configuration, add table.exec.mini-batch.enabled: true and table.exec.mini-batch.allow-latency: 2s.

3. Start the job.

   On the Status tab, you can see the MinibatchGlobalAggregate operator. Click + on the operator to verify the GlobalGroupAggregate operator has not sent data to downstream operators if the aggregation result remains unchanged.

   

4. Stop the job. On the Configuration tab, find the Parameters section, and click Edit. In Other Configuration, add table.exec.agg.mini-batch.output-identical-enabled: true.

5. Start the job.

   On the Status tab, you can see the MinibatchGlobalAggregate operator. Click + on the operator to check that the GlobalGroupAggregate operator sends data to downstream operators if the aggregation result remains unchanged.

Scenarios

• The number of messages with the same update key in a stream is small over a period of time. For example, the update key is regarded as the primary key and the data with the same primary key is not frequently updated. In addition, the processing order based on the update key is required when dimension tables are joined. In this case, you can perform optimization by specifying the table.exec.async-lookup.key-ordered-enabled parameter. This can ensure the data processing order based on the update key.

• In a Change Data Capture (CDC) stream that contains a primary key, dimension tables are joined to generate a wide table for data writing to the sink. The primary key of the sink is consistent with the primary key of the source. In addition, the join key of the JOIN operation on the dimension tables is inconsistent with the primary key. The join key is considered as the primary key. In this case, you can perform optimization by specifying the table.exec.async-lookup.key-ordered-enabled parameter. This allows the system to shuffle data based on the CDC primary key, which is regarded as the update key. You can also enable the SHUFFLE_HASH join policy to optimize this scenario. In high-concurrency scenarios, compared with this method, optimization by specifying the table.exec.async-lookup.key-ordered-enabled parameter can prevent the generation of the SinkMaterializer operator before data is written to the sink. This prevents potential performance issues caused by the operator, especially the issue that a large amount of state data is generated during the long-term running of the operator. For more information about the SinkUpsertMaterializer operator, see Usage notes.

• The join key of the JOIN operation on dimension tables is inconsistent with the primary key. The join key of the dimension tables is regarded as the primary key and a rank operator is available after the JOIN operation is performed. In this case, optimization by specifying the table.exec.async-lookup.key-ordered-enabled parameter allows the system to shuffle data based on the CDC primary key, which is regarded as the update key. You can also enable the SHUFFLE_HASH join policy to optimize this scenario. Compared with this method, optimization by specifying the table.exec.async-lookup.key-ordered-enabled parameter can prevent UpdateFastRank from degrading into RetractRank. For more information about how to change RetractRank to UpdateFastRank, see TopN practices.

Usage notes

• If no update key is available in a stream, the entire row of data is used as the key.

• The throughput decreases when the same update key is frequently updated in a short period of time. This is because the data with the same update key is processed in a strict order.

• Compared with JOIN operations on dimension tables in asynchronous mode before optimization, the Key-Ordered mode provides Keyed State. If you enable or disable the Key-Ordered mode, the state data compatibility is affected.

• The optimization takes effect only when you add the table.exec.async-lookup.output-mode='ALLOW_UNORDERED' and table.exec.async-lookup.key-ordered-enabled='true' configurations to the JOIN operation on a dimension table in VVR 8.0.10 or later and the input stream is an update stream.

Valid values

• false (default): disables the Key-Ordered mode.

• true: enables the Key-Ordered mode.

Example

1. Copy and paste the following code to a streaming draft and deploy it:

   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', -- The connector that supports the asynchronous lookup feature.
       'async' = 'true',
       'dbname' = 'holo db name', -- The name of the Hologres database.
       'tablename' = 'schema_name.table_name', -- The name of the Hologres table that is used 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 virtual private cloud (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. Go to the job details page. On the Configuration tab, find the Parameters section, and click Edit. In Other Configuration, add table.exec.async-lookup.output-mode='ALLOW_UNORDERED' and table.exec.async-lookup.key-ordered-enabled='true'.

3. Start the job.

   On the Status tab, you can view KEY_ORDERED:true in the async attribute of the job.

   

table.optimizer.window-join-enabled

This parameter controls whether to enable window joins. When it is enabled, Flink optimizes the execution plan based on the window join approach. For small windows, this can reduce state size and improve performance. Additionally, unlike regular joins, enabling window join can avoid sending update messages to downstream operators. This suits use cases that require joins over small windows.