[SUPPORT]Hudi Java client writes data very very slowly
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 116
Description
Every row data is about 0.72K. Every insert commit with 300 rows takes about 2 minutes. But increasing the rows of a commit, the time of every commit did not increase much. Event if each insert commit has 4500 rows, the time of every commit take just over 2 minutes.
1. Hudi version: 0.12.0, aliyun oss file system, flink 1.13.6.
2. Spark beeline is used to create the table. The table properties are as follows.
```
USING hudi
PARTITIONED BY (bizdate)
TBLPROPERTIES (
'primaryKey' = 'timestamp,serial_number,message_id',
'type' = 'mor',
'preCombineField' = 'timestamp',
'hoodie.datasource.write.hive_style_partitioning'='false',
'hoodie.database.name'='gs_ods',
'hoodie.table.base.file.format'='parquet',
'hoodie.parquet.writelegacyformat.enabled'='false'
);
```
3. Hudi Java client in a flink pipeline writes the data from a Kafka cluster, and the configuration is as follows.
```java
private String baseFileFormat = "parquet";
private String recordKeyFields;
private int parallelism = 20;
private WriteConcurrencyMode writeConcurrencyMode = WriteConcurrencyMode.SINGLE_WRITER;
private HoodieLockConfig hoodieLockConfig = HoodieLockConfig.newBuilder().build();
private HoodieFailedWritesCleaningPolicy hoodieFailedWritesCleaningPolicy
= HoodieFailedWritesCleaningPolicy.EAGER;
@Override
public void open(Configuration parameters) throws Exception {
****
// Create the write client to write some records in
HoodieWriteConfig cfg = HoodieWriteConfig.newBuilder()
.withPath(tablePath)
.withSchema(schema)
.forTable(tableName)
.withAutoCommit(true)
.withTableServicesEnabled(true)
.withEmbeddedTimelineServerEnabled(true)
.withMarkersType(MarkerType.TIMELINE_SERVER_BASED.name())
.withRollbackUsingMarkers(true)
.withDeleteParallelism(parallelism)
.withParallelism(parallelism, parallelism)
.withFinalizeWriteParallelism(parallelism)
.withRollbackParallelism(parallelism / 2)
.withWriteBufferLimitBytes(32 * 1024 * 1024)
.withWriteConcurrencyMode(writeConcurrencyMode)
.withLockConfig(hoodieLockConfig)
//.withEngineType(EngineType.SPARK)
.withCleanConfig(HoodieCleanConfig.newBuilder()
.withAutoClean(true)
.withFailedWritesCleaningPolicy(hoodieFailedWritesCleaningPolicy)
.withAsyncClean(false)
.build())
.withStorageConfig(
HoodieStorageConfig.newBuilder()
.parquetWriteLegacyFormat("false")
.build())
.withMetadataConfig(
HoodieMetadataConfig.newBuilder()
.withAsyncClean(false)
.withAsyncIndex(false)
.enable(true)
.build())
.withIndexConfig(
HoodieIndexConfig.newBuilder()
.withIndexType(IndexType.BLOOM)
.build())
.withArchivalConfig(HoodieArchivalConfig.newBuilder()
.archiveCommitsWith(40, 60)
.build())
.withCompactionConfig(
HoodieCompactionConfig.newBuilder()
.withCompactionLazyBlockReadEnabled(true)
.build())
.build();
client = new HoodieJavaWriteClient<>(new HoodieJavaEngineContext(hadoopConf), cfg);
}
@Override
public void invoke(T value, Context context) throws Exception {
List> records = toHoodieRecords(value);
if (null == records || records.size() == 0) {
return;
}
List statusList = null;
try {
String newCommitTime = client.startCommit();
statusList = client.insert(records, newCommitTime);
} catch (Exception e) {
log.error("Meet some errors " + Arrays.toString(records.toArray()), e);
throw e;
}
HashMap errors = statusList.get(0).getErrors();
if (null == errors || errors.size() == 0) {
return;
}
for (Map.Entry e : errors.entrySet()) {
log.error("Can not insert into " + e.getKey().getRecordKey(), e.getValue());
}
}
```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reproducing the reported timings with Hudi 0.12.0, Flink 1.13.6, and the listed OSS and write-client configuration. Inspect the HoodieJavaWriteClient startCommit and insert calls, comparing 300-row and 4,500-row commits; done means identifying the cause of the roughly two-minute commits and documenting a verified fix or configuration change.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100