You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "cbomgit (via GitHub)" <gi...@apache.org> on 2025/04/04 15:04:06 UTC
[I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
cbomgit opened a new issue, #13090:
URL: https://github.com/apache/hudi/issues/13090
**Describe the problem you faced**
We have a process that ingests 100 - 130GB of data per day into a COW table partitioned on year/month/day. The key is timestamp_eventId_skuNumber. A single batch may contain events that span up to 14 event dates (meaning we fan out to up to 14 partitions). Our write operation is an UPSERT in order to not have duplicates in the table.
For some input batches, we see very long index lookup times (3 - 10 hours), with large shuffle writes and reads (10 - 25 TB).
When I look at the tasks, it seems like each task is reading in a large amount of data, requiring us to create clusters with 1 executor per node. See the following task list:
When I look at the tasks that take the longest, I see the following types of log statements:
```
2025-04-04 04:35:01,928 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView: Sending request : (http://ip-10-0-133-87.ec2.internal:43393/v1/hoodie/view/datafile/latest/partition?partition=region%3DNA%2Fyear%3D2025%2Fmonth%3D03%2Fday%3D16&basepath=s3%3A%2F%2Fmads-ms-conversions-prod%2Fdaily-data%2Fglider&fileid=85460a75-3674-412c-8bec-36ea8a4345e7-0&lastinstantts=20250403234937872&timelinehash=30ed4d42429b272930e0456aba07ccac0e1b20f0cd421dab56afb2f1b6126952)
2025-04-04 04:35:01,967 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.io.HoodieKeyLookupHandle: Total records (7915696), bloom filter candidates (0)/fp(0), actual matches (0)
2025-04-04 04:35:01,967 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView: Sending request : (http://ip-10-0-133-87.ec2.internal:43393/v1/hoodie/view/datafile/latest/partition?partition=region%3DNA%2Fyear%3D2025%2Fmonth%3D03%2Fday%3D06&basepath=s3%3A%2F%2Fmads-ms-conversions-prod%2Fdaily-data%2Fglider&fileid=a9c2acd3-8dde-4d15-b0d8-458cd8847242-0&lastinstantts=20250403234937872&timelinehash=30ed4d42429b272930e0456aba07ccac0e1b20f0cd421dab56afb2f1b6126952)
2025-04-04 04:35:02,388 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem: Opening 's3://mads-ms-conversions-prod/daily-data/glider/region=NA/year=2025/month=03/day=06/a9c2acd3-8dde-4d15-b0d8-458cd8847242-0_2-2066-1095662_20250327160656670.parquet' for reading
2025-04-04 04:35:03,035 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.io.HoodieKeyLookupHandle: Read bloom filter from (region=NA/year=2025/month=03/day=06,a9c2acd3-8dde-4d15-b0d8-458cd8847242-0) in 1068 ms
2025-04-04 04:35:03,808 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView: Sending request : (http://ip-10-0-133-87.ec2.internal:43393/v1/hoodie/view/datafile/latest/partition?partition=region%3DNA%2Fyear%3D2025%2Fmonth%3D03%2Fday%3D06&basepath=s3%3A%2F%2Fmads-ms-conversions-prod%2Fdaily-data%2Fglider&fileid=a9c2acd3-8dde-4d15-b0d8-458cd8847242-0&lastinstantts=20250403234937872&timelinehash=30ed4d42429b272930e0456aba07ccac0e1b20f0cd421dab56afb2f1b6126952)
2025-04-04 04:35:03,847 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.io.HoodieKeyLookupHandle: Total records (17873), bloom filter candidates (0)/fp(0), actual matches (0)
2025-04-04 04:35:03,847 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView: Sending request : (http://ip-10-0-133-87.ec2.internal:43393/v1/hoodie/view/datafile/latest/partition?partition=region%3DNA%2Fyear%3D2025%2Fmonth%3D03%2Fday%3D09&basepath=s3%3A%2F%2Fmads-ms-conversions-prod%2Fdaily-data%2Fglider&fileid=df7becfe-2b4d-4e91-bc2d-e23f3bc91aa9-0&lastinstantts=20250403234937872&timelinehash=30ed4d42429b272930e0456aba07ccac0e1b20f0cd421dab56afb2f1b6126952)
2025-04-04 04:35:04,099 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem: Opening 's3://mads-ms-conversions-prod/daily-data/glider/region=NA/year=2025/month=03/day=09/df7becfe-2b4d-4e91-bc2d-e23f3bc91aa9-0_2-1502-791157_20250327143233103.parquet' for reading
2025-04-04 04:35:04,726 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.io.HoodieKeyLookupHandle: Read bloom filter from (region=NA/year=2025/month=03/day=09,df7becfe-2b4d-4e91-bc2d-e23f3bc91aa9-0) in 879 ms
2025-04-04 04:35:05,905 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView: Sending request : (http://ip-10-0-133-87.ec2.internal:43393/v1/hoodie/view/datafile/latest/partition?partition=region%3DNA%2Fyear%3D2025%2Fmonth%3D03%2Fday%3D09&basepath=s3%3A%2F%2Fmads-ms-conversions-prod%2Fdaily-data%2Fglider&fileid=df7becfe-2b4d-4e91-bc2d-e23f3bc91aa9-0&lastinstantts=20250403234937872&timelinehash=30ed4d42429b272930e0456aba07ccac0e1b20f0cd421dab56afb2f1b6126952)
2025-04-04 04:35:05,952 [INFO] (Executor task launch worker for task 0.0 in stage 32.0 (TID 66836)) org.apache.hudi.io.HoodieKeyLookupHandle: Total records (27262), bloom filter candidates (0)/fp(0), actual matches (0)
```
It seems to me like something isn't working correctly with our index, but I'm having trouble understanding what exactly is going on in order to help me tune our settings correctly. The following settings are being used for our job:
```
Computed insert options:
hoodie.datasource.write.insert.drop.duplicates -> false
hoodie.parquet.small.file.limit -> 104857600
hoodie.copyonwrite.record.size.estimate -> 106
hoodie.insert.shuffle.parallelism -> 12000
hoodie.datasource.write.precombine.field -> eventVersion
hoodie.bloom.index.filter.dynamic.max.entries -> 1270220
hoodie.cleaner.fileversions.retained -> 2
hoodie.parquet.max.file.size -> 134217728
hoodie.cleaner.parallelism -> 12000
hoodie.write.lock.client.num_retries -> 10
hoodie.bloom.index.prune.by.ranges -> true
hoodie.metadata.enable -> true
hoodie.clean.automatic -> false
hoodie.datasource.write.operation -> UPSERT
hoodie.write.lock.wait_time_ms -> 600000
hoodie.metrics.reporter.type -> CLOUDWATCH
hoodie.datasource.write.recordkey.field -> timestamp,eventId,trackedItem
hoodie.table.name -> ******
hoodie.datasource.write.table.type -> COPY_ON_WRITE
hoodie.datasource.write.hive_style_partitioning -> true
hoodie.write.lock.dynamodb.partition_key -> ******
hoodie.cleaner.policy -> KEEP_LATEST_FILE_VERSIONS
hoodie.write.markers.type -> DIRECT
hoodie.metrics.on -> false
hoodie.datasource.write.reconcile.schema -> true
hoodie.datasource.write.keygenerator.class -> org.apache.hudi.keygen.ComplexKeyGenerator
hoodie.cleaner.policy.failed.writes -> LAZY
hoodie.upsert.shuffle.parallelism -> 12000
hoodie.write.lock.dynamodb.table -> HoodieLockTable
hoodie.write.lock.provider -> org.apache.hudi.aws.transaction.lock.DynamoDBBasedLockProvider
hoodie.datasource.write.partitionpath.field -> region,year,month,day
hoodie.bloom.index.filter.type -> DYNAMIC_V0
hoodie.write.lock.wait_time_ms_between_retry -> 30000
hoodie.write.concurrency.mode -> optimistic_concurrency_control
hoodie.write.lock.dynamodb.region -> us-east-1
```
We're using a
**To Reproduce**
Steps to reproduce the behavior:
I'm unclear on how to fully reproduce, some batches of data ingest quickly (20 - 30 min), others take 3 - 10 hours. A couple of questions on my end are:
1. How can we better inspect the index behavior and verify it's doing the right thing?
2. Is there a way to log more verbose logs related to index behavior?
3. Would we be better off backing off to a SIMPLE BLOOM or just a SIMPLE INDEX?
**Expected behavior**
A clear and concise description of what you expected to happen.
**Environment Description**
* Hudi version : 11.1
* Spark version : 3.2.1
* Hive version : 3.1.3
* Hadoop version : 3.2.1
* Storage (HDFS/S3/GCS..) : s3
* Running on Docker? (yes/no) : no
--
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: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2797382093
@cbomgit I suggest you to 0.14.1 or 0.15.1 and try record index to optimize the index lookup stage.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "cbomgit (via GitHub)" <gi...@apache.org>.
cbomgit commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2779164719
Additional details:
I confirmed my approx record size by running a test with just a SIMPLE non-bloom index and ingesting a weeks worth of data. 128 MB files pack around 1.2M records per file, yielding around 95 bytes per record:
```
val df = spark.read.parquet("s3://path/to/table/region=NA/year=2025/month=03/day=18/*.parquet")
val countsDf = df.groupBy("_hoodie_file_name").agg(
countDistinct("_hoodie_record_key").as("count")
)
val percentiles = List(0.5, 0.75, 0.90, 0.95)
countsDf.agg(
avg("count").as("avg"),
min("count").as("min"),
max("count").as("max"),
percentile_approx($"count", typedLit(percentiles), lit(10000)).as("percentiles")
).select(
$"avg",
$"min",
$"max",
$"percentiles".getItem(0).as("p50"),
$"percentiles".getItem(1).as("p75"),
$"percentiles".getItem(2).as("p90"),
$"percentiles".getItem(3).as("p95")
).show(false)
val recordCountInFile = spark.read.parquet("s3://path/to/table/region=NA/year=2025/month=03/day=18/file_id.parquet").count
(121.0 * 1000 * 1000) / recordCountInFile
df: org.apache.spark.sql.DataFrame = [_hoodie_commit_time: string, _hoodie_commit_seqno: string ... 32 more fields]
countsDf: org.apache.spark.sql.DataFrame = [_hoodie_file_name: string, count: bigint]
percentiles: List[Double] = List(0.5, 0.75, 0.9, 0.95)
+------------------+------+-------+-------+-------+-------+-------+
|avg |min |max |p50 |p75 |p90 |p95 |
+------------------+------+-------+-------+-------+-------+-------+
|1263895.4036393713|224265|1333819|1266274|1267078|1267956|1268659|
+------------------+------+-------+-------+-------+-------+-------+
recordCountInFile: Long = 1265650 // records per file
res64: Double = 95.60304981629993 // record size
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2797370472
@cbomgit Can you upgrade hudi version, it sounds like very old.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "cbomgit (via GitHub)" <gi...@apache.org>.
cbomgit commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2779262374
Running a test using simple index seems to fair much better, though I am not yet sure how this will fair with our data deletion process. I'd love if I could better understand how to configure the dynamic bloom filter so that I can potentially have a more optimal configuration (or feel good about just using a Simple index).
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "cbomgit (via GitHub)" <gi...@apache.org>.
cbomgit commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2797429010
Hi @ad1happy2go , we may be able to try record index on this table (keys should be globally unique) - is there something specifically wrong with our current implementation or is it not the optimal use case? Or are we facing a bug or something else? Trying to understand what the current issue is and if it's possible to address it as it's quite difficult for us to upgrade versions at the moment.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] Facing long index lookup times with dynamic bloom filter [hudi]
Posted by "zjawj-123321 (via GitHub)" <gi...@apache.org>.
zjawj-123321 commented on issue #13090:
URL: https://github.com/apache/hudi/issues/13090#issuecomment-2816492997
开启/优化索引
Hudi spark 参数示例（推荐使用 Metadata Table Index）：
shell
--hoodie.metadata.enable=true
--hoodie.index.type=BUCKET
--hoodie.index.bucket.engine=CONSISTENT_HASHING
或尝试 Hudi 0.12+ 的 HFile Index 进行主键查找加速。
批次按分区拆分
输入数据先分组 by event_date，每次只 upsert 一到两天的数据。
定期合并小文件
使用 Hudi compaction/spark job，设置合并频率和目标文件大小。
shell
--hoodie.parquet.small.file.limit=134217728 # 128MB
写前去重
在 Spark 写入前：
scala
df = df.dropDuplicates("timestamp", "eventId", "skuNumber")
监控指标
监控表层级的文件数量、分区数据分布、索引查找时间、shuffle spill/GC 情况。
更多：https://code.mayoubang.cn/conversion/share/22805143081
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org