Hudi Table Streaming Reads

Updated on 2024-11-29 GMT+08:00

View PDF

Configurations

**Table 1** Configuration parameters for Hudi table streaming reads
Parameter	Description	Recommended Value	Mandatory
Connector	Type of the table to be read	hudi	Yes
Path	Path for storing the table	Set this parameter as needed.	Yes
table.type	Hudi table type. Available values are as follows: MERGE_ON_READ COPY_ON_WRITE (default value)	COPY_ON_WRITE	Yes
hoodie.datasource.write.recordkey.field	Primary key of the table	Set this parameter as needed.	Yes
write.precombine.field	Data combination field.	Set this parameter as needed.	Yes
read.tasks	Read task parallelism. The default value is 4.	4	No
read.streaming.enabled	Whether to enable incremental streaming read. The options are as follows: true: Incremental streaming read is enabled. false: Batch read is enabled.	true	Yes
read.streaming.start-commit	yyyyMMddHHmmss: Start commit time in yyyyMMddHHmmss format for incremental stream consumption. By default, the latest commit is used. The start and end commit time form a close interval. earliest The consumption starts from the beginning.	-	No
hoodie.datasource.write.keygenerator.type	Primary key generator type of the upstream table. The options are as follows: SIMPLE (default value) COMPLEX (When a Spark table is created, use this value, or set the value same as the one specified when the Spark table is created.) TIMESTAMP CUSTOM NON_PARTITION GLOBAL_DELETE	COMPLEX	No
read.streaming.check-interval	Check interval (in minutes) for detecting new upstream commits. Use the default value 1 when the traffic is heavy.	5	No
read.end-commit	End commit time. The start and end commit time form a close interval. By default, the latest commit time is used.	-	No
read.rate.limit	Traffic limit (records per second) for streaming read Default value 0 indicates that there is no limit. The value is the total limit. Limit for each operator = read.rate.limit/read.tasks (number of read operators).	-	No
changelog.enabled	Whether to write Changelog messages. The options are as follows: false: No Changelog message will be written (default value). true: Changelog messages will be written for CDC.	false	No
hoodie.datasource.write.hive_style_partitioning	Whether to use Hive style partitioning. The options are as follows: false: The Hive style is not used. Only the partition value (default value) is used as the partition directory name. true: The Hive style is used. The format of the partition directory name is <partition_column_name>=<partition_value>. If the Hudi partition table is created by Spark, this parameter is mandatory and must be set to true.	-	No

Development Suggestions

Set proper consumption parameters to avoid the "File Not Found" error.
When the downstream consumes Hudi files too slowly, the upstream archives the Hudi files. As a result, the "File Not Found" error occurs.

Optimization suggestions:
1. Improve the downstream read speed, for example, increase the number of read.tasks.
2. Improve the data consumption speed, for example, increase the traffic limit.
3. Increase the compaction period to prolong the retention period of log files.

Set the traffic limit for streaming read when a Hudi table is used as the source table.
If streaming read traffic exceeds the maximum traffic of the system, job exceptions may occur. Set a streaming read limit that equals the peak value verified by the service pressure test.

Run compaction on Hudi tables to prevent the long checkpointing of the Hudi Source operator.
If the Hudi Source operator takes long time for checkpointing, check whether the compaction of the Hudi table is normal. If compaction is not performed for a long time, the list performance deteriorates.
When streaming read is enabled for Hudi MOR tables, enable log indexing to improve the Flink streaming read performance.
The read and write performance of Hudi MOR tables can be improved through log indexing. Add 'hoodie.log.index.enabled'='true' for Sink tables and Source tables.
Impact of DDL changes on stream reading of Hudi tables
Adding columns using DDL has no impact on Hudi table reads. Other DDL changes (such as changing a column type and name, and deleting a column) affect Hudi table reads. Therefore, you need to stop the read jobs before changes are performed.