apache / apache/hudi

[SUPPORT] Incremental read query performance

Open
#10,334 2 comments 0 reactions 0 assignees View on GitHub
area:incr-processing area:performance engine:spark
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

**Describe the problem you faced**

I'm running an application that reads from 4 medium-sized (few hundred GB) Hudi MoR tables which are compacted weekly.

When running incremental queries to look over 3~5 commits with `hoodie.datasource.read.incr.fallback.fulltablescan.enable -> true`, I am seeing a 5 hour gap in my Spark History Server:
![image](https://github.com/apache/hudi/assets/24283126/8ffbb77a-79e3-4112-8d25-70069c5f2370)

Incremental read config:
```
Scan MergeOnReadIncrementalRelation(
org.apache.spark.sql.SQLContext@3b457d8a,Map(
hoodie.datasource.read.incr.fallback.fulltablescan.enable -> true,
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-10-80-36-225.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.read.begin.instanttime -> 20231207170214000,
hoodie.datasource.query.type -> incremental,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},None,None
)
Job time: 2023/12/12 00:47:50
=> 5 commits for incremental read
number of output rows: 19,926,780
```

During the gap, my driver kills all 40 of my executor instances and I see the following logs on driver: [log.txt](https://github.com/apache/hudi/files/13686484/log.txt)

When I use snapshot query + filter on _hoodie_commit_time, the gap reduces to ~20mins:
![image](https://github.com/apache/hudi/assets/24283126/138239ba-e2a8-4d3f-837c-727e4540c20d)

Snapshot read config:
```
Scan MergeOnReadSnapshotRelation(org.apache.spark.sql.SQLContext@24a4d185,
Map(
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-xx-xx-xxx.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.query.type -> snapshot,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),
HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},
List(),
None,
None
)
```

When I do incremental read with `hoodie.datasource.read.incr.fallback.fulltablescan.enable -> false`, I see the incremental query gap go from 5 hours -> 5 mins 😮 :
![image](https://github.com/apache/hudi/assets/24283126/3949a3c9-8b1c-485d-92d6-7b1be825ae16)

Incremental read config (no fulltable fallback):
```
Scan MergeOnReadIncrementalRelation(org.apache.spark.sql.SQLContext@dcb6efd,
Map(
path -> s3://lake/names.all_hudi,
hoodie.write.lock.zookeeper.url -> ip-10-xx-xx-xxx.ec2.internal,
hoodie.write.lock.zookeeper.base_path -> /hudi,
hoodie.metadata.enable -> true,
hoodie.datasource.hive_sync.jdbcurl -> jdbc:hive2://ip-10-xx-xx-xxx.ec2.internal:10000,
hoodie.datasource.read.begin.instanttime -> 20231212232646000,
hoodie.datasource.query.type -> incremental,
hoodie.cleaner.policy.failed.writes -> EAGER,
hoodie.write.lock.zookeeper.port -> 2181,
hoodie.write.lock.provider -> org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider,
hoodie.write.concurrency.mode -> single_writer),
HoodieTableMetaClient{basePath='s3://lake/names.all_hudi',
metaPath='s3://lake/names.all_hudi/.hoodie',
tableType=MERGE_ON_READ},
None,
None
)
number of output rows: 12,541,221
```

This leads me to believe the incremental query was degenerating into a full table scan when it shouldn't... (all my commits are active and uncleaned):
https://github.com/apache/hudi/blob/7a6543958368540d221ddc18e0c12b8d526b6859/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala#L142-L151

Wondering if someone can help me understand the following:
1. Why is the incremental load degenerating into a full table scan when I can run the same job with the flag disabled and do not run into File Not Found?
2. Is the full table scan expected to happen on only one instance (the driver?)
3. Can I safely leave this flag false, and expect to only run into FileNotFound if running cleaner?

**Expected behavior**

Only fall back to full table scan in incremental load when necessary (file not found issue) and not all the time.

**Environment Description**

* Hudi version : 0.13.1

* Spark version : 3.4.0

* Hive version : 3.1.3

* Hadoop version : 3.3.3

* Storage (HDFS/S3/GCS..) : S3

* Running on Docker? (yes/no) : no

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with MergeOnReadIncrementalRelation.scala at the referenced lines and compare the three incremental and snapshot configurations. Review log.txt alongside the full-table fallback setting and the reported executor shutdowns. Done means explaining when fallback is selected and whether the observed behavior matches the expected file-not-found-only fallback.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, scala
Domain
data-engineering, distributed-systems, performance
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.