Exchange prepared data with Ray or Spark

Keep one owner for each step

Your agent keeps the task goal, context and final decision. Ray or Spark owns local reads, transformations and resources; your application or a supported client prepares the declared file handoff. Lithi accepts the approved artifact reference and returns a result for your application to inspect. The framework does not become a Lithi scheduler.

Use Ray or Spark in your own environment to prepare input files and consume verified results. A file exchange is not a native cluster connector. The independent examples below use two synthetic records and make no Lithi request. Run each in a disposable development environment, not an existing shared cluster. Review your framework's local resource and telemetry settings separately.

Read a fixed Parquet file with Ray

This example requires Ray Data and PyArrow. It starts one local CPU worker allocation with a bounded object store and no dashboard. Ray documents initialization and read_parquet.

import os
from pathlib import Path
from tempfile import TemporaryDirectory
os.environ["RAY_USAGE_STATS_ENABLED"] = "0"
import pyarrow as pa
import pyarrow.parquet as pq
import ray
import ray.data
if ray.is_initialized():
    raise RuntimeError("Use a separate disposable Python process")
rows = [{"item_id": "sample-001", "text": "Hello"},
        {"item_id": "sample-002", "text": "Goodbye"}]
with TemporaryDirectory(prefix="lithi-ray-example-") as folder:
    path = Path(folder) / "records.parquet"
    pq.write_table(pa.Table.from_pylist(rows), path)
    ray.init(address="local", num_cpus=1, num_gpus=0,
             object_store_memory=100_000_000, include_dashboard=False)
    try:
        dataset = ray.data.read_parquet([str(path)], columns=["item_id", "text"])
        actual = sorted(dataset.take_all(), key=lambda row: row["item_id"])
        assert actual == rows
        print({"rows": len(actual)})
    finally:
        ray.shutdown()

Round-trip a local Spark dataset

This independent example requires PySpark and the Java runtime required by your Spark release. Use a separate disposable Python process. Spark's Parquet documentation covers its reader and writer behavior.

from pathlib import Path
from tempfile import TemporaryDirectory
from pyspark.sql import SparkSession
if SparkSession.getActiveSession() is not None:
    raise RuntimeError("Use a separate disposable Python process")
spark = (SparkSession.builder.master("local[1]")
         .appName("lithi-docs-synthetic")
         .config("spark.ui.enabled", "false").getOrCreate())
rows = [("sample-001", "Hello"), ("sample-002", "Goodbye")]
schema = "item_id string, text string"
try:
    with TemporaryDirectory(prefix="lithi-spark-example-") as folder:
        path = str(Path(folder) / "export")
        spark.createDataFrame(rows, schema).write.mode("errorifexists").parquet(path)
        actual = spark.read.schema(schema).parquet(path).orderBy("item_id").collect()
        assert [(row.item_id, row.text) for row in actual] == rows
        print({"rows": len(actual)})
finally:
    spark.stop()

Both examples report two rows. Their small collections are deliberate test fixtures, not a pattern for collecting a large production dataset into the driver.

Freeze production input files

Record exact source files, versions where available, byte counts, digests and the chosen schema. Preserve stable item IDs through repartitioning and filtering. Customer file partitions don't select Lithi execution placement. For Delta, Iceberg or Hudi tables, use the table-aware export procedure before ordinary Parquet readers.

Keep storage credentials in your framework's approved configuration. Use only a format accepted by the selected Compute workflow. Prepare the encrypted artifact through file preparation, then review availability and the quote before approving work. Don't submit a job from a retried Spark task or Ray transformation.

Reconcile results and recover

Verify accepted output before loading it. Preserve the job and result identities, inspect exceptions and join by item ID. Keep unknown, duplicate and missing IDs visible instead of silently dropping them.

If a framework task retries, resume its local read or transformation. If a destination write is uncertain, inspect its existing output before repeating it. Neither event independently approves another Compute job. If an input file changes or disappears, stop and review the source selection rather than accepting a partial dataset silently.

Next, review fixed table snapshots when your files come from a lakehouse.