dataproblems opened a new issue, #12828:
URL: https://github.com/apache/hudi/issues/12828
**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
### 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?
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.pdf](https://github.com/user-attachments/files/18759562/Details.for.Job.32.pdf)
[Details for Job
37.pdf](https://github.com/user-attachments/files/18759561/Details.for.Job.37.pdf)
[Details for Stage
70.pdf](https://github.com/user-attachments/files/18759560/Details.for.Stage.70.pdf)
[Details for Stage
98.pdf](https://github.com/user-attachments/files/18759558/Details.for.Stage.98.pdf)
[Spark Job View For
Upsert.pdf](https://github.com/user-attachments/files/18759559/Spark.Job.View.For.Upsert.pdf)
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
