Distributed Engine V1.5 (Flotilla) Roadmap

[colin-ho]

Overview

Distributed Engine V1.5 (Flotilla) is a redesign of Daft’s distributed Ray-based execution engine to improve scalability, performance, and observability, for multimodal data workloads.

Design doc: https://ripe-air-619.notion.site/Distributed-Engine-V1-5-Flotilla-1c9091bd8634801f812ef4896a156ed0

Development Milestones

🎯 Milestone 1: Foundation & Map-Only Operations ✅ COMPLETE

Objective: Establish core framework with map-only workloads and baseline observability

Core Operations Implemented

• Source
• Project
• ActorPoolProject
• Filter
• Sink
• Sample
• Explode
• Unpivot
• Limit

Observability Features

• Logging System
• Progress Bars

Key Workloads Optimized

• Batch Inference - ImageNet-scale model inference pipelines
• Document Processing - Common Crawl text processing workflows
• ETL Operations - High-throughput Parquet read/write operations (Extract-Load without Transform)

---

🚧 Milestone 2: Shuffle Operations IN PROGRESS

Objective: Enable shuffle-based operations using Ray object store

New Operations in Development

• Repartition
• Random Repartition
• Aggregate
• Distinct
• Window
• Hash Join
• Broadcast Join
• Sort
• Distributed Sort
• TopN
• Monotonically Increasing ID
• Concat
• Pivot
• Cross Join
• Into Partitions
• Sort merge join

Enhanced Observability

• Query-Level metrics - Export query level metrics like task completed / failed to subscribers

Target Workloads

• Document Deduplication - MinHash-based similarity detection for large document corpora
• TPCH Benchmark - Complete TPC-H decision support benchmark (all 22 queries)
• Complex Analytics - Multi-stage analytical pipelines with joins and aggregations

---

🎯 Milestone 3: Flight Shuffle & Enhanced Observability PLANNED

Objective: Achieve improve shuffle performance and scalability through Flight shuffle implementation, as well as node / task level observability

Flight Shuffle Implementation

• Default Flight Shuffle - Use Apache Arrow Flight for network communication
• Fault Tolerance - Robust handling of node failures during shuffle operations
• Performance Parity - Match or exceed current shuffle implementation

Production-Grade Observability

• Task-Level Monitoring - Individual task performance and resource usage
• Node-Level Metrics - Per-node CPU, memory, and disk utilization
• Performance Profiling - Detailed execution analysis and bottleneck identification

Key Workloads

• Large-Scale Repartitioning - Efficient data redistribution across cluster nodes

[colin-ho]

🚧 Milestone 2 TODOS:

• Test incremental repartitioning in local swordfish tasks perf: Concat partitions in flotilla repartition #4756
• Elide shuffle for groupbys if clustering spec of input matches the groupby keys feat(flotilla): GroupBy-style Ops #4699 (review)
• Elide local repartitions for groupbys (as the local swordfish groupby should be able to take care of the repartitioning) feat(flotilla): GroupBy-style Ops #4699 (review)
• Elide repartition for hash joins if inputs are already hash partitioned
• Elide repartition for distinct
• Remove oneshot channels from submittable task struct, instead construct when needed fix: flush pending task updates in HttpSubscriber on plan complete event #4736
• Refactor pipeline node to return either FinalizedTask (which needs to be materialized), or UnfinalizedTask (which can have more instructions added to it).
• Refactor pipeline nodes to be clearer on when materialization occurs.
• Refactor tasks and task lifecycle. Currently there is a lot of cofusion between pending task, scheduled task, submitted task, etc.
• Start all UDF actors across projections at the start
• Avoid all blocking ray.gets in flotilla, instead use await.
• Don't schedule empty scans (if all sources are empty), instead just produce an empty micropartition with the result schema. In general, let's try to special case on empty to reduce number of scheduled tasks.
• Create a distributed level chrome trace comprised of the chrome traces produced by the swordfish tasks
• Truncate task name in progress bar
• Auto-determine partitions for map-only workloads. For example, if we have a single metadata file containing urls for a url-download / large-udf operation, we should inject a repartition / into-partitions in order to parallelize properly.
• Auto-determine ‘concurrency’ on UDFs
• Unify resources shared across ‘swordfish tasks’ on the same worker (I/O clients, udf processes, memory)
• Improve flotilla shuffles. Currently they not as performant, there are many optimizations in the current ray runner that are not present (pre-shuffle merge, flight shuffle, parallel serialization, repartition optimizer rules)
• Configurable swordfish worker resource request
• Share scheduler across jobs

cc @srilman