apache / apache/texera

Design of Supporting Shuffling in Input Port Materialization Reader Threads

Open
#3,392 0 comments 0 reactions 1 assignee Claimed by @Xiao-zhen-Liu View on GitHub
design-decision engine evaluation refactor
Dominant language
Scala
Stars
314
Forks
187
Avg merge
1d 21h
Merged PRs (30d)
214

Description

### Background
Based on our offline discussions, we believe using **Input Port Materialization Threads** to Replace **Cache Read operators** makes the system cleaner mainly on the following two things:
- It is symmetric to the design of using output port materialization writer threads (#3295)
- It enables the scheduler to make its input plan **immutable** since we are no longer adding more operators during scheduling.

### Supporting Multiple Workers (Shuffling) for Input Port Materialization Reader Threads
A main complexity of this new design is where to put the shuffling logic in a cache read operator, since shuffling is necessary when there are multiple workers for one operator/input port.

We initially wanted to utilize Apache Iceberg's [hidden partitioning](https://iceberg.apache.org/docs/latest/partitioning/#icebergs-hidden-partitioning) feature, but further research and discussion revealed it is not currently possible to use Iceberg's partitioning for our shuffling (or any data system's shuffling).
Here is our conclusion about Iceberg Partitioning:
- Iceberg is used only as [source](https://iceberg.apache.org/docs/latest/spark-ddl/) and sink (result) operators in other data engines like Spark/Flink.
- The internal materializations during a Spark job execution currently have nothing to do with Iceberg:
- In the [early years of Spark](https://people.csail.mit.edu/matei/papers/2012/nsdi_spark.pdf) this materialization is stored as [local files of each worker](https://spark.apache.org/docs/latest/job-scheduling.html).
- Later Spark introduced [external shuffle service](https://spark.apache.org/docs/latest/job-scheduling.html#graceful-decommission-of-executors) to address issues of that previous simple design.
- Iceberg APIs are only concerned with partitioning and currently **only supports [hash](https://iceberg.apache.org/spec/#partition-transforms) / [range](https://www.datacamp.com/tutorial/apache-iceberg) based partitioning** (because these two are deterministic).
- If an Iceberg source/sink in a Spark job has a different partitioning scheme from the downstream/upstream operator, [a shuffling Spark operator will be introduced](https://iceberg.apache.org/docs/latest/spark-writes/#writing-distribution-modes) (similar to the cache read op we are trying to remove).
- **Shuffling** differs from **partitioning**, and for the current task of removing cache source, what we need to preserve is the shuffling logic of a cache source op (includes round-robin shuffling, broadcast shuffling, etc.), which is not achievable by directly using Iceberg Partitioning API.
- Texera/Amber differs from existing data systems in this sense because we rely on Iceberg to store both results and intermediate materializations.

### Conclusion

Our current simple design is to move the shuffling logic of a cache source op to the corresponding input port materialization reader thread
- Cons: This will introduce additional overhead because of duplicate read by multiple workers.
- Pros: Shuffling information on an input port will be part of the resource config of that port, which can be utilized by cost estimator and resource allocator because shuffling is an expensive resource

### Future Project
In the future, we may want to further optimize the shuffling on an input port, possibly by designing a partitioned materialization system (on top of Iceberg) that considers shuffling. This will be a separate project.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.