Realtime Compute for Apache Flink:Control state size to reduce backpressure in SQL deployments

Stateful operator types

SQL is a domain-specific and declarative language that allows you to perform data operations without the complexity of handling the underlying data processing logic. Flink SQL leverages the state backend and checkpointing mechanism of Apache Flink to ensure the final consistency of computation results. At the implementation level, Flink SQL utilizes an optimizer to select stateful operators based on parameter configurations and SQL statements. To optimize the performance of stateful computation over massive data, you need a basic understanding of the underlying mechanisms.

Stateful operators derived by the optimizer

The following table describes the stateful operators derived by the optimizer.

ChangelogNormalize

The ChangelogNormalize operator processes changelogs that involve primary keys and ensures efficiency, data consistency, and data accuracy. The operator is used in the following scenarios:

• The source table has primary keys and supports UPSERT operations.

  In this case, the source table is also referred to as an upsert source table. The table produces a changelog stream that contains only UPDATE (including INSERT and UPDATE_AFTER) and DELETE operations on the primary keys while maintaining the sequence of the primary keys. For example, you can use the Upsert Kafka connector to create an upsert source table. You can also override the getChangelogMode method to create a custom source connector that supports UPSERT operations.

  @Override
  public ChangelogMode getChangelogMode() {
      return ChangelogMode.upsert();
  }

• The 'table.exec.source.cdc-events-duplicate' = 'true' parameter is explicitly set.

  At-least-once processing for Change Data Capture (CDC) may generate duplicate changelog records. If you require exactly-once processing, specify this configuration to remove duplicate changelog records. Sample scenario:

  

  When this operator is present, it hash-shuffles the upstream data based on the primary key defined in the source table's DDL. The operator then uses ValueState to store the latest full-row record for each primary key. The following figure shows the process of updating the state and sending changes downstream. When processing the second -U(2, 'Jerry', 77), the state is already empty. This indicates that the +I/+UA and -D/-UB records have canceled each other out. The current retract message is a duplicate and can be discarded.

  

SinkUpsertMaterializer

The SinkUpsertMaterializer operator ensures that data materialization conforms to the upsert semantics when the sink table has a primary key. During stream data processing, if the uniqueness and sequence of the data records are disrupted before they are written to the sink table, the optimizer automatically adds this operator. The operator maintains its state based on the primary key of the sink table to ensure that the corresponding constraints are satisfied. For information about other common scenarios, see Handling out-of-order changelog events.

LookupJoin

When processing a LookupJoin operation, if you have explicitly configured the optimization option 'table.optimizer.non-deterministic-update.strategy'='TRY_RESOLVE' and the optimizer identifies a potential non-deterministic update issue (see Eliminating the impact of non-deterministic updates in streaming queries), the system attempts to resolve it. Specifically, if introducing a stateful operator can eliminate the non-determinism, the optimizer automatically creates a stateful LookupJoin operator.

Sample scenarios: (1) The primary keys of the sink table partially or completely overlap with those of the dimension table and data in the dimension table may be updated by using CDC or other tools. (2) The join operation involves a non-primary-key field in the dimension table. In the preceding scenarios, the LookupJoin operator can efficiently handle dynamic data changes while ensuring the accuracy and consistency of query results.

Stateful operators invoked by SQL statements

This type of stateful operator cleans up state data based on the TTL value or watermark progress. For example, stateful operators used for windowed computation, such as the WindowAggregate, WindowDeduplicate, WindowJoin, and WindowTopN operators, clean their state data based on the watermark progress. If the timestamp contained in a watermark is later than the end time of a window, the built-in timer triggers state cleanup.

L.time between R.time + X and R.time + Y 
  -- Or 
R.time between L.time - Y and L.time - X

Diagnostic tools

Backpressure is an indicator of performance bottlenecks in Apache Flink. In most cases, backpressure occurs because the state size continues to increase and exceeds the allocated memory size. In this case, the state backend moves infrequently used state data to the disk storage. However, accessing data in the disk storage is significantly slower than accessing data in the memory. If an operator frequently reads state data from the disk, the data latency significantly increases. This results in a performance bottleneck.

To identify whether backpressure is caused by a large state size, you need to thoroughly analyze the running status of the deployment and operators. For information about how to use the diagnostic tools of Realtime Compute for Apache Flink to identify performance issues, see Diagnostic tools.

Tuning methods

Avoid unnecessary use of stateful operators

This tuning method applies only to the stateful operators derived by the optimizer. In most cases, the stateful operators invoked by SQL statements are necessary.

• ChangelogNormalize

  This operator is invoked in most scenarios that involve an upsert source table, except for temporal joins based on event time. Before you use the Upsert Kafka connector or a similar connector, make sure that no temporal joins are performed based on event time. When the deployment is running, you need to monitor the state-related metrics of the ChangelogNormalize operator. If the source table has a large number of primary keys, the state size increases because the operator maintains its state based on primary keys (also referred to as keyed state). If the primary keys are frequently updated, the state data is frequently accessed and modified. We recommend that you do not use the Upsert Kafka connector to create a source table in scenarios such as data synchronization. We also recommend that you use a data synchronization tool that ensures exactly-once processing.

• SinkUpsertMaterializer

  The table.exec.sink.upsert-materialize parameter defaults to auto, which automatically checks for data consistency, especially if changelog events are out of order. This mechanism ensures data processing accuracy by introducing a SinkUpsertMaterializer. However, this operator's activation does not always indicate an out-of-order data issue. For example, when multiple group by keys are merged, the optimizer cannot accurately derive the upsert key. For safety, it adds a SinkUpsertMaterializer by default. If you have a thorough understanding of your data distribution and can ensure the correctness of the output without this operator, you can set the parameter to none. This setting guarantees both data correctness and performance.

  You can check whether the SinkUpsertMaterializer is active by inspecting the final node of your deployment's topology. In the execution graph, this operator is usually displayed together with the sink operator, forming an operator chain. This allows you to directly monitor and evaluate the use of SinkUpsertMaterializer in your data processing flow and make more informed optimization decisions.

  On the State Overview page of the deployment, the operator details table at the bottom shows the running status of each operator. A row like SinkMaterializer[952] -> Sink:... indicates that SinkUpsertMaterializer is active and chained with the Sink operator.

  If you detect that this operator is generated but the data computation is correct without it, set the parameter to 'table.exec.sink.upsert-materialize'='none' to prevent the SinkUpsertMaterializer from being added automatically. This parameter is not included in the default runtime parameter list. You must manually add it by clicking Add Custom Parameter on the deployment O&M page. For configuration steps, see How do I configure custom runtime parameters for a deployment?. Ververica Runtime (VVR) 8.0 and later versions introduce an intelligent SQL execution plan analysis feature to help you better identify such issues.

  In the DMS (Data Management Service) SQL editor, click the Deep Check button in the top toolbar. The system performs checks for SQL syntax correctness and network connectivity. After the check passes, the SQL Optimization area displays suggestions. For example, when a CONCAT expression is used to concatenate group keys as the sink table's primary key, the system alerts you that the declared primary key of the sink table does not match the UPSERT KEY inferred by the optimizer. It points out that CONCAT causes the loss of the UPSERT KEY property for the original fields (such as student_id and student_name) and provides recommendations to avoid implicit type conversions and conversions from numeric types to STRING.

  The SinkUpsertMaterializer operator stores its state data in Managed Memory and maintains one materialized record for each primary key. When a deployment has a large primary key cardinality with frequent updates, such as heartbeat data that contains high-frequency duplicate keys, the operator frequently creates Java objects and writes them to the managed memory. This puts continuous allocation pressure on the JVM heap memory and triggers frequent garbage collection (GC).

  If the heap memory configured for a TaskManager is insufficient, the frequent object creation can cause a JVM out-of-memory (OOM) error (Java heap space). This may also trigger Gemini state backend errors (Gemini DB error) because the Gemini state backend depends on the JVM environment.

  To resolve this issue, use one of the following solutions:

  • If you have verified data correctness, set 'table.exec.sink.upsert-materialize'='none' to disable the operator and eliminate the memory pressure. This parameter is not included in the default runtime parameter list. You must manually add it by clicking Add Custom Parameter on the deployment O&M page. For configuration steps, see How do I configure custom runtime parameters for a deployment?.

  • If your business logic requires the SinkUpsertMaterializer operator, increase the heap memory allocated to the TaskManager to ease the JVM heap memory pressure.

Reduce state access frequency: Enable miniBatch

If a minute-level latency is acceptable in your business, you can enable miniBatch to reduce the frequency of state access and updates. MiniBatch-related parameters are not included in the default runtime parameter list. You must manually add them by clicking Add Custom Parameter on the deployment O&M page. For more information, see Optimize Flink SQL.

The following table describes the stateful operators that support miniBatch in Realtime Compute for Apache Flink.

Reduce state size: Specify proper TTL values

State size is crucial to performance. To control the state size of a deployment, configure the State Expiration Time parameter based on your business requirements on the Deployments page in the development console of Realtime Compute for Apache Flink. For more information, see Parameters.

An excessively short TTL may lead to incorrect computation results. For example, if data arrives late and the relevant state data expires during an aggregation or join operation, the result is incorrect. An excessively long TTL increases resource consumption and reduces stability. We recommend that you configure the TTL based on the data characteristics and your business requirements. For example, if your daily data computations have a maximum drift of 1 hour across days, you can set the TTL to 25 hours.

VVR 8.0.1 and later allow you to use the JOIN_STATE_TTL hint to specify different TTL values for the states of the left and right streams in a regular join. This reduces unnecessary state storage and improves performance. For information about how to use the hint, see Hints.

SELECT /*+ JOIN_STATE_TTL('left_table' = '..', 'right_table' = '..') */ *
FROM left_table [LEFT | RIGHT | INNER] JOIN right_table ON ...

The following table describes the state sizes of a deployment before and after the JOIN_STATE_TTL hint is used.

Reduce state size: Optimize execution plan

The optimizer selects the state implementation and generates an execution plan based on the SQL statements and specified configurations.

• Use primary keys to optimize regular joins

  • If the join keys contain primary keys, the system uses a ValueState<RowData> object to store only the most recent value of each join key. This maximizes storage space utilization.

  • If the join keys contain non-primary keys, the system uses a MapState<RowData, RowData> object to store the most recent record from the source table based on the primary key for each join key.

  • If no primary key is defined, the system uses a MapState<RowData, Integer> object to store the entire data record corresponding to each join key and the number of times the data record appears.

  To optimize storage efficiency, we recommend that you define primary keys in the DDL statements of the tables that you want to join and perform a regular join based on the primary keys.

• Optimize deduplication over append-only streams

  Use the ROW_NUMBER function to replace the FIRST_VALUE or LAST_VALUE function to increase deduplication efficiency. When you use the ROW_NUMBER function to obtain the first or most recent data record, the Deduplicate operator stores only the first or most recent record of the specified keys.

• Improve aggregation performance

  Use the FILTER syntax to replace the CASE WHEN syntax in multi-dimensional data aggregation, such as when you want to calculate the number of unique visitors on mobile devices, desktop devices, and all devices. The SQL optimizer can recognize the different filter arguments on the same key. This allows the state data to be shared when multiple COUNT DISTINCT values are calculated based on different conditions on the same key. This also reduces the number of times the state data is accessed. Tests show that the FILTER syntax can provide twice the performance of the CASE WHEN syntax.

Reduce state size: Adjust the join order of multiple streams

Apache Flink uses binary hash joins to process data streams. In the following example, the join of streams A and B results in unnecessary storage consumption. The issue becomes more significant as the number of joined streams increases.

To address this issue, adjust the joining order. For example, you can join a stream that has a smaller amount of data before a stream that has a larger amount of data. This helps mitigate the amplification effect caused by state redundancy and improves the efficiency and performance of data processing.

Minimize disk reads

You can reduce the number of times to access disk storage to improve system performance. You can also optimize memory utilization. For more information, see Minimize disk reads.

References

• For information about the issues caused by a large state size and the tuning workflow, see Large-state jobs.

• Apache Flink Datastream API allows you to manage the state size in a flexible manner. For more information, see Reduce backpressure in DataStream jobs.

• For information about how to diagnose and prevent checkpoint and savepoint timeout, see Diagnose and tune checkpoint and savepoint timeouts

• For information about how to identify and remove performance bottlenecks during deployment startup and scaling, see Improve startup and scaling speed.