Introduction
The SET statement can adjust the key runtime parameters of a job. Currently, most parameters can be configured via the Job Advanced Parameters feature. Only a very small number of parameters require the use of the SET statement. Note:
SET is an advanced statement. Use it with caution to avoid exceptions caused by improper parameter configuration during runtime.
The SET command does not support comments. Do not add -- comment information after it.
A semicolon is required at the end of a SET statement.
Syntax
In a SET statement, string-type values should be quoted with single quotation marks. This is not necessary for Boolean or numeric values.
SET configuration item = 'parameter value';
Examples
Configuring the idle state retention time for an SQL job
For statements with large state such as GROUP BY and JOIN, Flink provides the Idle State Retention Time mechanism. You can configure the minimum and maximum state retention times to prevent unbounded state growth from causing job crashes (OOM). For example, the following statement sets the minimum state retention time to 3 hours.
Note:
The minimum and maximum retention times should be at least 5 minutes apart. Otherwise, an error will occur and Flink will ignore the settings.
The format of time units should follow the configuration requirements of Flink. For example, you can set the time to 10min, 3h, 3hour, 7day, or 7d. Spaces between numbers and time units are optional.
set table.exec.state.ttl = '3h';
Enabling Mini-Batch Support for Tables
Flink SQL provides Mini-Batch support for aggregations, which can significantly improve throughput. This feature is disabled by default because it increases processing latency. To use Mini-Batch, add the following statements on the SQL job editing page to enable this feature through the following settings. The batch size and latency parameters can be customized but cannot be omitted.
set table.exec.mini-batch.enabled = true;
set table.exec.mini-batch.size = 5000;
set table.exec.mini-batch.allow-latency = '200 ms';
Avoiding the Use of SinkUpsertMaterializer
SinkUpsertMaterializer maintains the full quantity in its state, which may cause the state to become excessively large.
You can configure this through table.exec.sink.upsert-materialize:
auto (default): Flink automatically determines whether out-of-order records may exist. If so, SinkUpsertMaterializer is added.
none: not used.
force: force the use of SinkUpsertMaterializer.
Note that the SinkUpsertMaterializer automatically added by Flink may not always be necessary. You can disable it through the following configuration if the job parallelism is 1 or you know that the join Key will not change.
table.exec.sink.upsert-materialize = 'none';