[SUPPORT]Loss record when complete compaction
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 112
Description
```shell
df = spark.read.format("hudi").load("/user/hive/warehouse/test.db/test_table")
df.filter("id = '4d9254de8b14aa536fc6c3c2f3ee5341'").select("ts", "first_seen", "event_time", 'indicator', "_hoodie_record_key", "id").show(truncate=False)
+---+----------+----------+---------+------------------+---+
|ts |first_seen|event_time|indicator|_hoodie_record_key|id |
+---+----------+----------+---------+------------------+---+
+---+----------+----------+---------+------------------+---+
df = spark.read.format("hudi").load("/user/hive/warehouse/test.db/test_table/*.parquet")
df.filter("id = '4d9254de8b14aa536fc6c3c2f3ee5341'").show(truncate=False)
|_hoodie_record_key |id |_hoodie_commit_time|_hoodie_partition_path|_hoodie_file_name |_hoodie_commit_seqno |
+------+---------+-------------+-------------+-------------+-----------+--------------------------------+--------------------------------+-------------------+----------------------+------------------------------------------------------------------------+-----------------------------+
|4d9254de8b14aa536fc6c3c2f3ee5341|4d9254de8b14aa536fc6c3c2f3ee5341|20221229102613056 | |00000085-a43f-4bc2-9ad8-3580d7385530-0_38-0-38_20231015061029133.parquet|20221229102613056_85_57980048|
+------+---------+-------------+-------------+-------------+-----------+--------------------------------+--------------------------------+-------------------+----------------------+------------------------------------------------------------------------+-----------------------------+
```
**Expected behavior**
hudi snapshot mode query MOR table, no records found, but found in all versions of parquet files,
use spark structed streaming, read from kafka and write hudi, async compact,
Does this record loss phenomenon exist in version 0.12.1 and which version is fixed?
pls help!!!
```
```python
hudi write config
keep_max_file_version = 12
keep_min_commits = 10000
keep_max_commits = keep_min_commits + 1
hudi_options = {
'hoodie.table.name': hive_table,
'hoodie.datasource.write.recordkey.field': "id",
'hoodie.datasource.write.table.name': hive_table,
'hoodie.datasource.write.operation': "upsert",
'hoodie.datasource.write.precombine.field': "ts",
'hoodie.datasource.write.table.type': "MERGE_ON_READ",
'hoodie.upsert.shuffle.parallelism': 300,
'hoodie.bulkinsert.shuffle.parallelism': 2000,
'hoodie.insert.shuffle.parallelism': 2000,
"hoodie.compact.inline": "false",
"hoodie.compact.inline.max.delta.commits": 1,
"hoodie.compact.schedule.inline": "true",
'hoodie.cleaner.policy': 'KEEP_LATEST_FILE_VERSIONS',
'hoodie.cleaner.fileversions.retained': keep_max_file_version,
"hoodie.keep.min.commits": keep_min_commits,
"hoodie.keep.max.commits": keep_max_commits,
'hoodie.parquet.max.file.size': 1024 * 1024 * 100,
'hoodie.parquet.small.file.limit': 1024 * 1024 * 60,
'hoodie.parquet.compression.codec': 'snappy',
# 'hoodie.datasource.write.partitionpath.field': "dt",
'hoodie.datasource.hive_sync.partition_extractor_class': 'org.apache.hudi.hive.NonPartitionedExtractor',
'hoodie.datasource.write.keygenerator.class': "org.apache.hudi.keygen.NonpartitionedKeyGenerator",
'hoodie.datasource.write.hive_style_partitioning': "true",
'hoodie.datasource.hive_sync.enable': 'false',
'hoodie.datasource.hive_sync.database': db,
'hoodie.datasource.hive_sync.table': table,
'hoodie.datasource.hive_sync.mode': "hms",
"hoodie.index.type": "BUCKET",
"hoodie.bucket.index.num.buckets": 100,
"hoodie.index.bucket.engine": "SIMPLE", # CONSISTENT_HASHING
"hoodie.storage.layout.partitioner.class": "org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner",
'hoodie.datasource.write.payload.class': "org.apache.hudi.common.model.GtHoodieRecordPayload",
}
```
**Environment Description**
* Hudi version : 0.12.1
* Spark version : 3.2.2
* Hive version : 2.1.1
* Hadoop version : 3.0.0
* Storage (HDFS/S3/GCS..) : HDFS
* Running on Docker? (yes/no) : no
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```Add the stacktrace of the error.```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reproducing the supplied Spark snapshot read against the MOR table after asynchronous compaction, using the listed Hudi 0.12.1, Spark 3.2.2, and HDFS environment. Compare the snapshot result with the direct parquet-path query and the configured cleaner and compaction settings; done means determining whether the record loss is reproducible and identifying the version containing a fix.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- hadoop, kafka, python, spark
- Domain
- data-engineering, databases, distributed-systems, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100