apache / apache/hudi

[SUPPORT]Hudi Java client writes data very very slowly

Open
#6,585 5 comments 0 reactions 0 assignees View on GitHub
area:performance area:writer priority:high
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.