apache / apache/hudi

[SUPPORT] Hudi Upsert operation taking too long under writing data for base table and metadata table

Open
#12,828 1 comment 0 reactions 0 assignees View on GitHub
area:performance
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

**Describe the problem you faced**

I am trying to perform an upsert to hudi with a dataframe of 200 M records and I noticed that it is taking an hour to complete this process. My hudi table has record level index enabled on it and I would like to speed up this upsert process to finish in a matter of minutes as opposed to an hour. Please see the attached UI screenshots specifically jobs 32 and 37 and under them stages 70 and 98 respectively.

We are upserting about `181309872` records to the table. I see that there are `23360` occurrences of fileId in the commit file, which I'm assuming is the number of files that hudi modifies during this upsert operation.

### Questions ###

1. For stage 98, where does the 20003 number for tasks come from? None of my spark config has that value and I want to know if I can configure it so that I can try increasing it to get a lower time for updating the metadata.
2. For stage 70, is there anything that we can do to improve the time it takes to write the data files? We've to use a copy on write table for our consumers to get the latest version after the commit file. It would be good to understand if there are any configurations that we can tune to get a better performance here.
3. Should we include any other configuration in our upsert configuration to ensure that our data upsert works in the most efficient manner?
4. Another puzzling thing I observed was the size of the commit files for upsert operations - they were almost as large as the commit file that was created for the table creation operation using INSERT? Do you know why that might be? See the following image for the sizes.

![Image](https://github.com/user-attachments/assets/6744aa4b-63a5-49a2-9873-27038c5699d6)

I'm trying to determine if we have hit the lower limit for the performance in terms of time of upsert operation and see if there is room for improvement or not.

**To Reproduce**

* Read the data
* Upsert into hudi

**Expected behavior**

The upsert operation would finish in minutes as opposed to taking an hour.

**Environment Description**

* Hudi version : 0.15.0

* Spark version : 3.4.1

* Hive version :

* Hadoop version : 2.7.5

* Storage (HDFS/S3/GCS..) : S3

* Running on Docker? (yes/no) : no

### Spark Configuration ###
```
"spark.checkpoint.compress": "true",
"spark.default.parallelism": "15800",
"spark.driver.cores": "8",
"spark.driver.defaultJavaOptions": "-XX:+UseG1GC -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseStringDeduplication -Dlog4j2.configuration=log4j2.properties",
"spark.driver.maxResultSize": "0",
"spark.driver.memory": "53g",
"spark.driver.memoryOverhead": "10g",
"spark.eventLog.logBlockUpdates.enabled": "true",
"spark.executor.cores": "8",
"spark.executor.defaultJavaOptions": "-XX:+UseG1GC -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseStringDeduplication -Dlog4j2.configuration=log4j2.properties",
"spark.executor.memory": "53g",
"spark.executor.memoryOverhead": "10g",
"spark.hadoop.mapred.output.compress": "true",
"spark.hadoop.mapred.output.compression.codec": "snappy",
"spark.io.compression.codec": "zstd",
"spark.memory.fraction": "0.80",
"spark.memory.storageFraction": "0.30",
"spark.num.executors": "395",
"spark.rdd.compress": "true",
"spark.scheduler.reporterThread.maxFailures": "5",
"spark.serializer": "org.apache.spark.serializer.KryoSerializer",
"spark.shuffle.compress": "true",
"spark.shuffle.spill.compress": "true",
"spark.sql.adaptive.coalescePartitions.parallelismFirst": "false",
"spark.sql.adaptive.enabled": "true",
"spark.sql.autoBroadcastJoinThreshold": "20965760",
"spark.sql.files.maxPartitionBytes": "536870912",
"spark.sql.inMemoryColumnarStorage.compressed": "true",
"spark.sql.parquet.fs.optimized.committer.optimization-enabled": "true",
"spark.sql.shuffle.partitions": "15800",
"spark.storage.level": "MEMORY_AND_DISK_SER"
```

### Hudi upsert configuration ###

We disable the timeline server and use DIRECT markers as per [this other support ticket](https://github.com/apache/hudi/issues/12116#issuecomment-2419613375)

```
hoodie.embed.timeline.server -> false,
hoodie.parquet.small.file.limit -> 1073741824,
hoodie.metadata.record.index.enable -> true,
hoodie.datasource.write.precombine.field -> $precombineField,
hoodie.datasource.write.payload.class -> org.apache.hudi.common.model.OverwriteWithLatestAvroPayload, hoodie.metadata.index.column.stats.enable -> true,
hoodie.parquet.max.file.size -> 2147483648,
hoodie.metadata.enable -> true,
hoodie.index.type -> RECORD_INDEX,
hoodie.datasource.write.operation -> upsert,
hoodie.parquet.compression.codec -> snappy,
hoodie.datasource.write.recordkey.field -> $recordKey,
hoodie.table.name -> $tableName,
hoodie.datasource.write.table.type -> COPY_ON_WRITE,
hoodie.datasource.write.hive_style_partitioning -> true,
hoodie.cleaner.policy -> KEEP_LATEST_COMMITS,
hoodie.write.markers.type -> DIRECT,
hoodie.populate.meta.fields -> true,
hoodie.datasource.write.keygenerator.class -> org.apache.hudi.keygen.SimpleKeyGenerator,
hoodie.cleaner.policy.failed.writes -> LAZY,
hoodie.upsert.shuffle.parallelism -> 23700,
hoodie.cleaner.commits.retained -> 10,
hoodie.datasource.write.partitionpath.field -> $partitionField
```

### Spark UI screenshots with details

#### Details for Job 32

![Image](https://github.com/user-attachments/assets/58f34153-587f-45f4-b80b-e9aa58141264)

#### Details for Job 37

![Image](https://github.com/user-attachments/assets/a4d51371-6d30-4968-88c4-49149e67caac)

#### Details for Stage 70

![Image](https://github.com/user-attachments/assets/7d4c7fe4-0bc7-45a3-8b25-769b16575b17)

#### Details for Stage 98

![Image](https://github.com/user-attachments/assets/1065c5a0-96c4-45f8-87de-1e9b54cbc3ea)

#### Spark Job View for the upsert operation

![Image](https://github.com/user-attachments/assets/a69c092d-586e-46b2-b853-21b56dec703e)

### Hudi table partition with object count and size ###

We tried using a partition key that logically partitions the data, however, due to the nature of our data we have a skew in our partitions as you can see.
![Image](https://github.com/user-attachments/assets/febd387f-5381-4c2b-b08f-436861ba5dd4)

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the Spark UI details for stages 70 and 98 and the listed Hudi upsert configuration, especially the record index and shuffle parallelism settings. Determine what controls the task count, data-file writing time, metadata update time, and commit-file size; done means providing an evidence-based explanation or a reproducible performance fix.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, performance
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.