RayOrch: Programming and Executing Lineage-Controlled Multi-Grain Dataflows for Foundation-Model Data Preparation

1 Introduction

• Long-tailed, input-dependent expansions make efficient cross-parent batching physically useful while requiring each child’s parent, ordinal, and terminal outcome for correct gathers and failure containment.Expansion membership and ordinal order, not batch composition or completion order, determine reconstruction.Source: §1 · p. 1
• Coarse per-source functions hide child-level scheduling, while flat interfaces leave membership and terminal outcomes to applications and can require post-stage regrouping.Those applications must define parent completion and failure semantics themselves.Source: §1 · pp. 1–2
• RayOrch retains runtime structural lineage: concrete ordered child membership, immediate parents, immutable ordinals, and terminal outcomes, with expansion and gather relations declared and compiler-validated.Structural lineage is online scheduling and completion state, not post hoc provenance.Source: §1 · p. 2
• F.expand declares and materializes ordered child Entities, while matching F.reduce gathers them by ordinal; per-Call FIFO queues batch ready Grains across parents, and typed GroupFailure suppresses same-parent undispatched siblings.Unrelated parents remain live, and RayOrch supports ordered gathering within each source microbatch, not general joins or cross-microbatch windows.Source: §1 · p. 2
• 13.1% lower end-to-end time than Ray Data and 29.0% than Daft on MinerU, plus 16.0% lower than Ray Data on Docling; scaling reaches 15.14× on MinerU and 7.82× on video.The results are on NVIDIA H20 GPUs.Source: §1 · p. 2

2.1 Multimodal Data Preparation

• Document parsing yields input-dependent pages and regions, while each root video yields an input-dependent clip sequence; retain each derived unit’s immediate-parent association and sequence position when accelerator batches mix sources.Text-curation pipelines such as Llama 3 and FineWeb instead emphasize staged filtering and deduplication over predominantly flat document records [13] [27].Source: §2.1 · p. 3

2.2 Pipeline Execution Systems

• Distributed frameworks expose complementary execution layers, from task graphs and streaming batches to multimodal dataframes and data-curation or training-data infrastructure.The cited systems include Spark, Dask, Ray Data [19], Daft, Data-Juicer [8], Mixtera [5], and Modyn.Source: §2.2 · p. 3
• RayOrch makes ordered variable-cardinality expansions part of its program and runtime contract, maintaining membership and child lineage to drive ordered gathers and same-parent failure containment.Physical batches may mix Grains from different parents.Source: §2.2 · p. 3

2.3 Structure, Progress, and Provenance

• Prior work preserves nested query structure, tracks temporal or iterative progress, or records provenance; RayOrch instead uses a materialized parent’s ordered child sequence for online scheduling and gathering.These systems track distinct state: nested collections [30], logical timestamps, event-time progress, and data-level provenance.Source: §2.3 · p. 3

3 System Overview

• RayOrch presents a logical view of data units and structural transformations alongside an execution view that schedules computations on available workers.The logical structure remains stable as batching, retries, and worker placement change.Source: §3 · pp. 3–4, Figure 2 · p. 4

3.1 Logical Dataflow

• A Domain names a hierarchy level, an Entity is one unit at that level, and applying a configured Call to an Entity forms a Grain, the unit of computation.A Port carries logical values, with one Item for each Port and Entity.Source: §3.1 · p. 4
• F.expand exposes an ordered child list as child Entities, and F.reduce reconstructs the parent Item in declared order from the program’s relation rather than rediscovering groups from record fields.Users can express page- or frame-level parallelism without manually maintaining child counts, grouping state, or sorting logic.Source: §3.1 · p. 4

3.2 Physical Execution

• Each configured Call has a per-Call FIFO Ready Queue and worker pool; ready Grains can be packed into transient batches that mix parents without making those batches logical groups.A parent’s next-stage work can start once its declared children are complete, without waiting for unrelated parents.Source: §3.2 · p. 4

4.1 Compiling the Declarative Structural Plan

• The compiler checks structural primitives and Domain compatibility, then lowers them into an immutable graph whose concrete child Entities and Grains are still produced at runtime.The graph contains Calls, Ports, Domains, consumers, and pre-indexed structural effects.Source: §4.1 · pp. 4–5

4.2 Lineage State and Parent Completion

• Lineage facts record each expansion’s terminal outcome, while input propagation and ordinal-ordered reduction determine whether a parent completes, drops, or is suppressed.Dropped members are excluded; present values are emitted in immutable ordinal order, and failed or suppressed members suppress the parent.Source: §4.2 · p. 5

4.3 Scheduling Ready Grains

• Per-Call FIFO queues dispatch ready Grains under reservation policies that change physical batch composition without changing Grain identity or reconstruction order.Elastic reservation may mix parents; parent-bound reservation selects siblings of the first ready parent. Downstream Calls dispatch independently after committed outcomes, without a stage-wide regrouping barrier.Source: §4.3 · p. 5, Figure 3 · p. 6

4.4 Commit, Failure, and Recovery

• Typed GroupFailure suppresses queued siblings within one Call and parent, while in-flight siblings cannot commit; untyped execution failures may instead trigger retries or actor replacement.Already committed siblings and unrelated parents remain unaffected; generation checks reject stale or duplicate reports.Source: §4.4 · pp. 5–6

4.5 Semantic Contract

• For fixed per-Grain outcomes, legal physical schedules yield the same final Items and lineage order, though payload-byte equivalence additionally requires UDF invariance to batch shape, input order, and randomness.Schedules may differ in batch composition, worker placement, or retry timing.Source: §4.5 · p. 6

• RayOrch runs as a Python layer on Ray, with persistent actor pools per configured Call and the driver retaining the compiled plan, lineage metadata, and pending references.Payloads remain in Ray’s object store; physical actor handles and references do not enter logical Grain identity.Source: §5 · p. 6
• RayModule stages implement batch UDFs that receive and return corresponding lists of values, while RayOrch manages lineage, batching, completion, and instrumentation.Workers validate declared arity and lengths and return per-Grain reports; UDFs do not maintain parent IDs or ordering.Source: §5 · p. 6
• Adapters cover MinerU and Docling document stages and video pipelines using Qwen2.5-VL-7B between decoding and per-source merge.Persistent actor instances amortize initialization of models and other heavyweight resources across batches.Source: §5 · p. 6

6 Evaluation

• The evaluation asks how performance and scalability compare, what explains gains, whether FIFO helps beyond rebatching, and whether lineage suppresses doomed sibling work without losing healthy outputs.Source: §6 · pp. 6–7

6.1 Experimental Setup

• Evaluation covers MinerU (3,689 PDFs), video (27,091 videos), and Docling (2,000 PDFs), using NVIDIA H20 GPUs and shared valid inputs and model pipelines within each workload.GPU counts range from 4–64 for MinerU, 8–64 for video, and 4 for Docling.Source: §6.1 · p. 7, Table 4 · p. 7

6.2 RQ1: Performance and Scalability

• At 64 GPUs, MinerU reaches 15.14× speedup and 94.6% of ideal scaling; its 4,295.7-second E2E time is 13.1% below Ray Data and 29.0% below Daft.These results come from separate strong-scaling and 64-GPU end-to-end comparisons.Source: Figure 5 · p. 7, Table 5 · p. 7, §6.2 · p. 7

6.3 RQ2: Why RayOrch Gains

• RayOrch overlaps OCR with assembly and upload, completing parents locally instead of leaving post-OCR regrouping tails.Stage windows overlap; they are not additive, and all work remains included in the E2E measurement.Source: §6.3 · pp. 7–8

6.4 RQ3: Scheduling Ablations

• FIFO scheduling reduces wall time from 634.1 to 579.3 seconds, an 8.6% reduction beyond rebatching.The controlled comparison holds lineage and commit fixed while adding FIFO to rebatching.Source: Table 7 · p. 8, §6.4 · p. 8

6.5 RQ4: Lineage-Scoped Failure Containment

• In controlled failure injection, typed containment suppresses 6,241 ready siblings and cuts paired wall time by 14.93% mean while preserving all expected healthy outputs.The 6,241 suppressed siblings are 26.5% of poisoned-parent siblings and 13.7% overall.Source: Table 8 · p. 8, §6.5 · p. 8

7 Conclusion

• RayOrch combines compiler-validated expansions and parent-scoped gathers with runtime lineage, FIFO queues, generation-fenced commits, and typed failure containment for finite, acyclic multi-grain dataflows.Source: §7 · p. 8
• On NVIDIA H20 GPUs, RayOrch reduces MinerU E2E time by 13.1% versus Ray Data and 29.0% versus Daft, and Docling E2E time by 16.0% versus Ray Data.Its strong-scaling speedups through 64 GPUs are 15.14× on MinerU and 7.82× on video.Source: §7 · p. 8
• FIFO scheduling reduces wall time by 8.6%, while lineage-scoped containment suppresses 6,241 non-trigger sibling computations and preserves expected outputs for unaffected parents.Source: §7 · p. 8