[SUPPORT] Zordering clustering on a moderate size dataset taking large amounts of time.

[vinothchandar]

Describe the problem you faced

I am trying to play with z-ordering on a 50G+ dataset locally to understand everything. Noticed large number of stages, and its pretty slow due to that. I want to make sure this is expected.

To Reproduce

Steps to reproduce the behavior:

1. Any 50GB+ dataset. I am using the amazon reviews dataset here https://s3.amazonaws.com/amazon-reviews-pds/readme.html
2. Run inline compaction

val df = spark.read.parquet(inputPath)
val commonOpts = Map("hoodie.bulk_insert.shuffle.parallelism" -> "10",
                     "hoodie.clustering.inline" -> "true",
                     "hoodie.clustering.inline.max.commits" -> "1",
                     "hoodie.layout.optimize.enable" -> "true",
                     "hoodie.clustering.plan.strategy.sort.columns" -> "product_id,customer_id,review_date")
df.write.format("hudi").
  option(PRECOMBINE_FIELD.key(), "review_id").
  option(RECORDKEY_FIELD.key(), "review_id").
  option("hoodie.table.name", "amazon_reviews_hudi").
  option(OPERATION.key(),"bulk_insert").
  option(BULK_INSERT_SORT_MODE.key(), "NONE").
  options(commonOpts).
  mode(Overwrite).
  save(outputPath)

Expected behavior

A clear and concise description of what you expected to happen.

Environment Description

• Hudi version : 0.10-SNAPSHOT

• Spark version : Apache Spark 3.0

• Hive version :

• Hadoop version :

• Storage (HDFS/S3/GCS..) : Local filesystem

• Running on Docker? (yes/no) :

Additional context

[vinothchandar]

@xiarixiaoyao Just want your eyes on this to see if this is expected. (It's possible I am doing this from older commit, not latest master with #4060 ). I ll post updated screenshots if needed, as soon as I can get them

[xiarixiaoyao]

@vinothchandar
The cluster operation splits too many jobs, and each job will be zorder. Spark supports concurrent jobs. Your program is not concurrent enough

By the way, clustering itself is slow.

In the first version of zorder code, we introduced the optimize operation for z-order, which was fast, but the community did not agree with this method, so I deleted it

[xiarixiaoyao]

let me run you code on my cluster。

[vinothchandar]

@xiarixiaoyao yes I have only 4 cores. Lets file a JIRA under our umbrella to make this fast going forward! we can work together on that

[xiarixiaoyao]

@vinothchandar
Thank you for supporting this work. The benchmark results I posted before are run through optimize， not by clustering， clustering is too slow since of its own mechanism. I suggest that we introduce z-order / Hilbert in SQL, just like databrick

[xiarixiaoyao]

@vinothchandar
I added a new issue to track this problem。 https://issues.apache.org/jira/browse/HUDI-2873

[rubenssoto]

Hello @xiarixiaoyao and @vinothchandar

I am testing clustering for a demonstration and I decide to reproduce the Vinoth test on my cluster.

Machine type: r5.4xlarge
Number of nodes: 8

Same dataset as Vinoth
s3://amazon-reviews-pds/parquet/

hudi_options = {
'hoodie.table.name': 'amazon_reviews_hudi',
'hoodie.datasource.write.recordkey.field': 'review_id',
'hoodie.datasource.write.precombine.field': 'review_id',
'hoodie.datasource.write.table.name': 'amazon_reviews_hudi',
'hoodie.datasource.write.operation': 'bulk_insert',
'hoodie.datasource.write.keygenerator.class': 'org.apache.hudi.keygen.NonpartitionedKeyGenerator',
'hoodie.parquet.small.file.limit': '805306368',
'hoodie.parquet.max.file.limit': '1073741824',
'hoodie.parquet.block.size': '805306368',
'hoodie.metadata.enable': 'true',
"hoodie.bulk_insert.shuffle.parallelism": "30",
"hoodie.clustering.inline": "true",
"hoodie.clustering.inline.max.commits": "1",
"hoodie.layout.optimize.enable":"true",
"hoodie.clustering.plan.strategy.sort.columns": "product_id,customer_id,review_date"
}

The job took 23 minutes.

I think the z order jobs are not running in parallel.

Is it possible to launch this jobs in different threads?
createOptimizedDataFrameByMapValue at RDDSpatialCurveOptimizationSortPartitioner.java:83

for better resource utilization.

[xiarixiaoyao]

@rubenssoto Thank you for your feedback
createOptimizedDataFrameByMapValue at RDDSpatialCurveOptimizationSortPartitioner.java:83 should support running in parallel， let me check it thanks

[xiarixiaoyao]

@rubenssoto @vinothchandar i have run the test。
suggestions： if you use z-order/hilbert in clustering, pls set option("hoodie.clustering.plan.strategy.max.bytes.per.group", Long.MaxValue.toString)
we need as many files as possible participate in sorting，in this way the sorting effect will be the best and there is no problem of parallelism

On the issue of parallelism, the current mechanism of cluster itself makes it impossible to do z-sort in parallel
Let me submit a PR to solve this problem

[rubenssoto]

Thank you @xiarixiaoyao

I will test with bigger values on hoodie.clustering.plan.strategy.max.bytes.per.group.
Only one more question, I want to demonstrate data skipping to my teammates, how can I enable data skipping on reading side on pyspark?

[xiarixiaoyao]

@rubenssoto
two ways
1）：spark.sql("set hoodie.enable.data.skipping = True")
2）：df = spark.read.format("hudi").option("hoodie.enable.data.skipping", "True").load("/tmp/default/hilbert")
then use sql/dataFrame to query

[xiarixiaoyao]

@vinothchandar
The current cluster mechanism just can't support concurrency very well. Even if you use ordinary sorting (not z-order / Hilbert), there also exsit this problem. Let me think about how to modify this mechanism， this is the pr #4178

[rubenssoto]

@xiarixiaoyao I follow your tip about hoodie.clustering.plan.strategy.max.bytes.per.group and use your patch from pr 4178 and the job now took 15 minutes, it is a good improvement.

Both tasks were submited at the same time.

[xiarixiaoyao]

@rubenssoto
Thank you for your test. We will continue to optimize this area in the future。

Give me a little advice：
if the sort columns has skewed， you can try set。
LAYOUT_OPTIMIZE_CURVE_BUILD_METHOD.key -> "sample"
see the benchmark result of org.apache.spark.sql.execution.benchmark.SpaceCurveOptimizeBenchMark