Realtime Compute for Apache Flink:Hudi connector

Parameters for parallelism

Parameters for online compaction

File-related parameters

File-related parameters are used to manage the file size. The following table describes the supported parameters.

Hadoop-related parameter

Parameters for data writing

Hudi supports various write methods, including batch write and streaming write. Hudi supports various data types, including changelog data and log data. Hudi also supports different indexing schemes.

• Parameters for batch write

  If you want to import existing data from other data sources to Hudi, you can use the batch import feature to import existing data into a Hudi table.

  Parameter	Description	Default value	Remarks
  write.operation	The type of the write operation.	upsert	Valid values: upsert: Data is inserted and updated. insert: Data is inserted. bulk_insert: Data is written in batches. Note If you set this parameter to bulk_insert, Avro-based data serialization and data compaction are not performed. Deduplication is not performed after data is imported. If you have high requirements on the uniqueness of data, make sure that the imported data is unique. You can set the write.operation parameter to bulk_insert only in batch execution mode. In this mode, the system sorts input data by partition name and writes the data to a Hudi table by default. This way, the write operation is not frequently switched among different files and the system performance is not degraded.
  write.tasks	The parallelism of bulk_insert write tasks.	Parallelism of Flink deployments	The parallelism of bulk_insert write tasks is specified by the write.tasks parameter and has an impact on the number of small files. In theory, the parallelism of bulk_insert write tasks is equal to the number of buckets. If data that is written to a file in each bucket reaches the maximum size of the file, data is rolled over to a new file handle. Therefore, the final number of write files is greater than or equal to the parallelism of bulk_insert write tasks. The maximum size of a Parquet file is 120 MB.
  write.bulk_insert.shuffle_input	Specifies whether to shuffle input data based on partition fields for batch insert tasks.	true	In Hudi 0.11.0 and later, you can set this parameter to true to reduce the number of small files. However, this may cause data skew.
  write.bulk_insert.sort_input	Specifies whether to sort input data based on partition fields for batch insert tasks.	true	In Hudi 0.11.0 and later, if you want to run a write task to write data to multiple partitions, you can set this parameter to true to reduce the number of small files.
  write.sort.memory	The available managed memory of the sort operator.	128	Unit: MB.

• Parameter for enabling the changelog mode

  In changelog mode, Hudi retains all changes (INSERT, UPDATE_BEFORE, UPDATE_AFTER, and DELETE) of messages and then works together with the stateful computing of the Flink engine to implement end-to-end near-real-time data warehousing. MOR tables of Hudi support the retention of all changes of messages based on row-oriented storage. Fully managed Flink can read an MOR table in streaming mode to consume all change records.

  Note

  In non-changelog mode, the intermediate changes in a batch dataset in a single streaming read can be merged. All immediate results are merged in a batch read (snapshot read). The intermediate state is ignored regardless of whether it is written.

  Parameter	Description	Default value	Remarks
  changelog.enabled	Specifies whether to consume all changes.	false	Valid values: true: All changes can be consumed. false: Not all changes can be consumed. The UPSERT semantics is used. Only the last merged message among all messages is ensured, and the intermediate changes may be merged. Note After you set the changelog.enabled parameter to true, the asynchronous compaction task still merges the intermediate changes into one data record. If data is not consumed at the earliest opportunity during streaming read, only the last data record can be read after data is compacted. You can change the buffer time for data compaction to reserve a specific time for the reader to read and consume data. For example, you can set the compaction.delta_commits parameter to 5 and the compaction.delta_seconds parameter to 3600.

• Append mode (supported in Hudi 0.10.0 and later)

  In Append mode:

  • The small file policy is applied to MOR tables. Data is written to Avro log files in append mode.

  • The small file policy is not applied to COW tables. A new Parquet file is generated each time data is written to a COW table.

Parameters for clustering policies

Hudi supports various clustering policies to resolve the small file issue in INSERT mode.

• Parameter for inline clustering (supported only for COW tables)

  Parameter	Description	Default value	Remarks
  write.insert.cluster	Specifies whether to merge small files during data writing.	false	Valid values: true: Small files are merged during data writing. false: Small files are not merged during data writing. Note By default, small files are not merged when the INSERT operation is performed on a COW table. If you set this parameter to true, the existing small files are merged each time the INSERT operation is performed. However, deduplication is not performed. As a result, the write throughput decreases.

• Parameters for async clustering (supported in Hudi 0.12.0 and later)

  Parameter	Description	Default value	Remarks
  clustering.schedule.enabled	Specifies whether to schedule the clustering plan during data writing.	false	If you set this parameter to true, the clustering plan is periodically scheduled.
  clustering.delta_commits	The number of commits that are required to generate a clustering plan.	4	If the clustering.schedule.enabled parameter is set to true, this parameter takes effect.
  clustering.async.enabled	Specifies whether to asynchronously run the clustering plan.	false	If you set this parameter to true, the clustering plan is asynchronously run at regular intervals to merge small files.
  clustering.tasks	The parallelism of clustering tasks.	4	N/A.
  clustering.plan.strategy.target.file.max.bytes	The maximum size of a file for clustering.	1024 * 1024 * 1024	Unit: bytes.
  clustering.plan.strategy.small.file.limit	The file size threshold for small files that are used for clustering.	600	Only files whose size is less than the value of this parameter can be used for clustering.
  clustering.plan.strategy.sort.columns	Columns based on which data is sorted for clustering.	No default value	You can specify special sorting fields.

• Parameters for clustering plan strategies

  Parameter	Description	Default value	Remarks
  clustering.plan.partition.filter.mode	The partition filter mode that is used in the creation of a clustering plan.	NONE	Valid values: NONE: Partitions are not filtered. All partitions are selected for clustering. RECENT_DAYS: If data is partitioned by day, the partitions that correspond to the specified number of recent days are selected for clustering. SELECTED_PARTITIONS: The specified partitions are selected for clustering.
  clustering.plan.strategy.daybased.lookback.partitions	The number of recent days based on which partitions are selected for clustering when the clustering.plan.partition.filter.mode parameter is set to RECENT_DAYS.	2	This parameter takes effect only when the clustering.plan.partition.filter.mode parameter is set to RECENT_DAYS.
  clustering.plan.strategy.cluster.begin.partition	The start partition, which is used to filter partitions.	No default value	This parameter takes effect only when the clustering.plan.partition.filter.mode parameter is set to SELECTED_PARTITIONS.
  clustering.plan.strategy.cluster.end.partition	The end partition, which is used to filter partitions.	No default value	This parameter takes effect only when the clustering.plan.partition.filter.mode parameter is set to SELECTED_PARTITIONS.
  clustering.plan.strategy.partition.regex.pattern	The regular expression that is used to specify partitions.	No default value	N/A.
  clustering.plan.strategy.partition.selected	The selected partitions.	No default value	Separate multiple partitions with commas (,).

• Parameters related to the bucket index type

  Note

  The parameters in the following table are supported in Hudi 0.11.0 and later.

  Parameter	Description	Default value	Remarks
  index.type	The index type.	FLINK_STATE	Valid values: FLINK_STATE: The Flink state index type is used. BUCKET: The bucket index type is used. If the amount of data is large such as more than 500 million data records in a table, the storage overhead of Flink state may become a bottleneck. The bucket index type uses a fixed hash policy to allocate data that contains the same key to the same file group. This helps prevent the storage and query overheads of the index. The bucket index type and the Flink state index type have the following differences: Compared with the Flink state index type, the bucket index type does not have storage and computing overheads. The bucket index type provides better performance than the Flink state index type. If you use the bucket index type, you cannot increase the number of buckets. If you use the Flink state index type, you can dynamically increase the number of files based on the file size. The bucket index type does not support cross-partition changes. If the input data is CDC streaming data, this limit is not applicable. The Flink state index type does not have limits.
  hoodie.bucket.index.hash.field	The hash key field when the bucket index type is used.	Primary key	This parameter can be set to a subset of the primary key.
  hoodie.bucket.index.num.buckets	The number of buckets when the bucket index type is used.	4	By default, the value of this parameter is the number of buckets in each partition. You cannot change the value of this parameter after the configuration.

Parameters for data reading

• Hudi supports various read methods, including batch read, streaming read, and incremental data read. Hudi can also consume and transfer changelogs to implement end-to-end incremental ETL.

  • Parameters for streaming read

    By default, snapshot reading is used for tables. In this case, the latest full snapshot data is read and returned at a time. You can set the read.streaming.enabled parameter to true to enable streaming read. You can configure the read.start-commit parameter to specify the start offset for streaming read. You can set the read.start-commit parameter to earliest to allow data to be consumed from the earliest offset.

    Parameter	Description	Default value	Remarks
    read.streaming.enabled	Specifies whether to enable streaming read.	false	Valid values: true: Streaming read is enabled. false: Streaming read is disabled.
    read.start-commit	The start offset for streaming read.	Left empty	Valid values: Time in the yyyyMMddHHmmss format: Data is consumed from the specified time. earliest: Data is consumed from the earliest offset. If this parameter is left empty, data is consumed from the latest time.
    clean.retain_commits	The maximum number of historical commits that can be retained by the cleaner.	30	If the number of historical commits exceeds the value of this parameter, the excess historical commits are deleted. In changelog mode, this parameter can be used to control the period for which changelogs can be retained. For example, if the checkpointing period is 5 minutes, changelogs are retained for at least 150 minutes by default. Important Streaming read of changelogs is supported only in Hudi 0.10.0 and later. In changelog mode, Hudi retains changelogs for a period of time for downstream consumers to consume. Changelogs may be merged in compaction tasks. In this case, intermediate records are deleted, which may affect the calculation result.

  • Parameters for incremental data read (supported in Hudi 0.10.0 and later)

    Fully managed Flink supports incremental consumption by calling a DataStream API, batch incremental consumption, and batch consumption of data at a specific point in time by using the time travel feature.

    Parameter	Description	Default value	Remarks
    read.start-commit	The offset from which data consumption starts.	Commit from the latest offset	The values of these parameters are in the yyyyMMddHHmmss format. The interval is a closed interval, which contains the start and end offsets.
    read.end-commit	The offset at which data consumption ends.	Commit from the latest offset