[SUPPORT] Compaction & Clustering are not working

[Cpandey43]

Describe the problem you faced

Issue:1
I configured the application with async compaction, async clustering, and async cleaning in the job but all are not working as per the configured settings.

async compaction - Not working at all
async clustering - Not working at all
async cleaning - cleaning was executed after every commit

Issue:2
Configured table _type is MOR with operation upsert but in partitions, Hudi is not creating .log files and also generating 18MB parquet files. Application created two partitions and both have maximum 18MB files. Attaching the content of both partitions below FYR.

Environment Description

• Hudi version : 0.13.1

• Spark version : 3.2.4

• Storage (HDFS/S3/GCS..) : Minio as a S3 extension

• Running on k8s : Using spark operator

Hudi conf used in spark application

    df.write
    .format("org.apache.hudi")
    // Write Config
    .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
    .option("hoodie.datasource.write.precombine.field", "ts")
    .option("hoodie.datasource.write.recordkey.field", "recordkey")
    .option("hoodie.datasource.write.partitionpath.field", "date")
    .option("hoodie.datasource.write.table.name", "spark_streaming")
    .option("hoodie.table.name", "spark_streaming")
    .option("hoodie.datasource.write.operation", "upsert")
    .option("hoodie.merge.small.file.group.candidates.limit", "1")
    // Hive Sync
    .option("hoodie.datasource.hive_sync.mode", "hms")
    .option("hoodie.datasource.hive_sync.metastore.uris", "thrift://hive-metastore:9090")
    .option("hoodie.datasource.hive_sync.database", "hudi_test")
    .option("hoodie.datasource.hive_sync.table", "test_table")
    .option("hoodie.datasource.hive_sync.partition_fields", "date")
    .option("hoodie.datasource.hive_sync.enable", "true")
    // Compaction
    .option("hoodie.compact.inline", "false")
    .option("hoodie.compact.inline.max.delta.commits", "6")
    .option("hoodie.datasource.compaction.async.enable", "true")
    .option("hoodie.parquet.small.file.limit", "104857600")
    // Clustering
    .option("hoodie.clustering.async.enabled", "true")
    .option("hoodie.clustering.async.max.commits", "1")
    .option("hoodie.clustering.execution.strategy.class", "org.apache.hudi.client.clustering.run.strategy.SparkSortAndSizeExecutionStrategy")
    // Cleaning
    .option("hoodie.clean.async", "true")
    .option("hoodie.cleaner.commits.retained", "3")
    // Archive
    .option("hoodie.archive.async", "true")
    // Index
    .option("hoodie.index.type", "BLOOM")
    // Payload
    .option("hoodie.payload.event.time.field", "ts")
    .option("hoodie.payload.ordering.field", "ts")
    // KeyGenerator
    .option("hoodie.datasource.write.hive_style_partitioning", "true")
    .option("hoodie.datasource.write.keygenerator.class", "org.apache.hudi.keygen.SimpleKeyGenerator")
    // Marker
    .option("hoodie.rollback.using.markers", "true")
    // Multi-writes
    .option("hoodie.write.concurrency.mode", "optimistic_concurrency_control")
    .option("hoodie.write.lock.provider", "org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider")
    .option("hoodie.write.lock.zookeeper.url", "zookeeper")
    .option("hoodie.write.lock.zookeeper.port", "2181")
    .option("hoodie.write.lock.zookeeper.base_path", "/hoodie")
    .option("hoodie.cleaner.policy.failed.writes", "LAZY")
    .option("hoodie.write.lock.zookeeper.lock_key", "lock_v1")
    .option("hoodie.write.lock.zookeeper.connection_timeout_ms", "30000")
    .option("hoodie.write.lock.wait_time_ms", "600000")
    // Multi-Modal Index 
    .option("hoodie.metadata.index.bloom.filter.enable", "true")
    .option("hoodie.metadata.index.column.stats.enable", "true")
    // MetaData 
    .option("hoodie.metadata.enable", "true")
    // Data Skipping
    .option("hoodie.enable.data.skipping", "true")
    // Storage
    .option("hoodie.parquet.max.file.size", "125829120")
    .option("hoodie.logfile.max.size", "1073741824")
    .option("hoodie.logfile.to.parquet.compression.ratio", "0.35"))

hoodie.properties file content

#Updated at 2023-11-21T20:26:57.745347Z
#Tue Nov 21 20:26:57 UTC 2023
hoodie.compaction.payload.class=org.apache.hudi.common.model.OverwriteWithLatestAvroPayload
hoodie.table.type=MERGE_ON_READ
hoodie.table.metadata.partitions=bloom_filters,column_stats,files
hoodie.table.precombine.field=ts
hoodie.table.partition.fields=date
hoodie.archivelog.folder=archived
hoodie.table.cdc.enabled=false
hoodie.timeline.layout.version=1
hoodie.table.checksum=3323849328
hoodie.datasource.write.drop.partition.columns=false
hoodie.table.timeline.timezone=LOCAL
hoodie.table.name=spark_streaming
hoodie.table.recordkey.fields=recordkey
hoodie.compaction.record.merger.strategy=eeb8d96f-b1e4-49fd-bbf8-28ac514178e5
hoodie.datasource.write.hive_style_partitioning=true
hoodie.partition.metafile.use.base.format=false
hoodie.table.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.populate.meta.fields=true
hoodie.table.base.file.format=PARQUET
hoodie.database.name=
hoodie.datasource.write.partitionpath.urlencode=false
hoodie.table.version=5

.hoodie folder content captured in below attached file name hoodie.txt
hoodie.txt

.hoodie folder recursive all folders content captured in below-attached file name hoodie-recursive.txt
hoodie-recursive.txt

Application processed two days data 1st & 2nd November. Attaching the content of both partitions below:
date=2023-11-01.txt
date=2023-11-02.txt

[noahtaite]

@Cpandey43

Hey! I'm another Hudi 0.13.1 MOR user, so just thought I'd come by to help lend a hand and dig a bit deeper into the problem you're reporting. To be clear - I'm not a Hudi developer.

First question - are you ingesting any updated records or only new records? From what I understand, the expected behaviour for MoR table (which I can confirm you are using based on your configurations + properties) is:

• new records are written to base .parquet files
• updates to records are written to avro .log files

New records need to get associated with a parquet file before updates get logged to avro files. I'm not a Hudi developer but to me it seems like this is the case partially so that a read optimized query will correctly show new records in the data set. However updates will not show until compaction is ran against those log files. It is how I've always understood the tradeoff here.

Second question - what is the behaviour you expect to see with async clustering? It should indeed be a "no-operation" until you asynchronously schedule + execute clustering to stitch together small files / improve query performance by having control over data locality. I suggest taking a look at the following guide here and appropriate 0.13.1 configurations to see what suits your needs:

hoodie.clustering.async.enabled
hoodie.clustering.inline
hoodie.clustering.schedule.inline

Out of the box these are all disabled, so we wouldn't see any clustering actions in your timeline. Also, be aware that when you do use clustering, you would see a ".replacecommit" in your Hudi timeline.

I will leave the cleaner questions to someone more familiar with this Hudi table service.

[Cpandey43]

@noahtaite Thanks for your support!!

Issue: 1
Even for updates in the same partition, Hudi is not generating .log files.

Issue: 2
I'm expecting Hudi will create new thread for scheduling & executing clustering without disturbing the write process.
I've configured below setting in my code & expecting others clustering config will pick default values.

hoodie.clustering.async.enabled - true
hoodie.clustering.async.max.commits - 1
hoodie.clustering.execution.strategy.class -org.apache.hudi.client.clustering.run.strategy.SparkSortAndSizeExecutionStrategy

Also, any idea on why it is creating base files of 18MB?

[subash-metica]

@Cpandey43 I had the same questions. The name async is misleading, currently this suggests the current writer to not perform the specific table service (clustering, compaction) inline. Rather, to inform the writer that async service will be scheduled and/or executed independently. Hence, this config doesn't run the table service as a daemon (in a separate thread), you had to perform the operation as a separate job.

I managed to run a separate table services job with distributed locking enabled (using ZooKeeper) in order to avoid write conflicts.

[Cpandey43]

@Cpandey43 I had the same questions. The name async is misleading, currently this suggests the current writer to not perform the specific table service (clustering, compaction) inline. Rather, to inform the writer that async service will be scheduled and/or executed independently. Hence, this config doesn't run the table service as a daemon (in a separate thread), you had to perform the operation as a separate job.

I managed to run a separate table services job with distributed locking enabled (using ZooKeeper) in order to avoid write conflicts.

@vinothchandar @nsivabalan @codope Can you please confirm this one??

[nsivabalan]

hey @ad1happy2go : can you follow up on this.
@Cpandey43 : yes you are right. enabling async w/ batch writers like spark-ds does not mean much.