Gowtham Potureddi

Posted on Sep 3

Daft: A Rust-Backed Distributed DataFrame for Multimodal & ML Data

#python #sql #interview #dataengineering

Daft is the DataFrame that finally stops forcing you to choose between a tool that fits in your head and a tool that fits your data — a Python-first API sitting on a Rust execution engine that stays lazy, plans and optimizes your query, and then runs it either multithreaded on your laptop or distributed across a cluster from the same code. The hard problem was never "transform a table"; it was that the two dominant options each fail a different half of the modern workload. Pandas is friendly and expressive but single-node and eager, so it falls over the moment the data outgrows memory. Spark scales to petabytes but drags a JVM behind it, speaks Python through a serialization boundary, and — the part that matters most in 2026 — cannot represent an image, a tensor, or an embedding as a genuine column, so every machine-learning pipeline ends up bolting a pile of hand-written glue onto the side of the DataFrame.

This guide is the engineer's walkthrough of the tool built for exactly that gap — a distributed DataFrame whose columns can hold multimodal values (URLs, bytes, images, tensors, embeddings) as first-class typed data, and whose engine was written in Rust so the heavy work never crosses back into slow Python per-row code. It is framed the way a strong platform interview probes it: why lazy evaluation plus a query optimizer beats eager row-by-row execution, how a columnar Arrow representation lets the Rust engine stream larger-than-memory data out-of-core, how one program runs locally or on Ray by flipping a runner, how multimodal columns and UDFs turn download-decode-embed into a few expressions, and how the whole thing reads Parquet and Iceberg from object storage, runs batch inference, and feeds a training loop. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the data processing practice library →, rehearse pipeline shaping on the data transformation practice library →, and sharpen the architecture axis with the system design practice library →.

On this page

Why Daft — the gap it fills
The execution model — lazy DataFrame, optimizer, columnar
Distributed execution on Ray — local to cluster
Multimodal columns — URLs, images, embeddings
ML data pipelines — Parquet, Iceberg, batch inference
Cheat sheet — Daft for multimodal & ML data
Frequently asked questions
Practice on PipeCode

1. Why Daft — the gap it fills

The consumption gap — Pandas is single-node, Spark can't hold an image, and ad-hoc Ray re-invents a query engine

The one-sentence invariant: Daft is a Python-first DataFrame on a Rust engine that is lazy, columnar, and multimodal-native, and it exists because the modern workload — big, and full of images, tensors, and embeddings — falls into the seam between Pandas (expressive but single-node and eager), Spark (scalable but JVM-bound and tabular-only), and Polars (fast but single-machine), so Daft's whole job is to give you one API that keeps a laptop program correct on a cluster, treats media as first-class typed columns, and pushes the heavy work into Rust so it never bottlenecks on per-row Python. Reach for Pandas and the data outgrows the machine; reach for Spark and you cannot put a decoded image in a column; hand-roll a Ray script and you have quietly signed up to maintain a worse query engine than the one Daft already ships.

The four axes interviewers actually probe.

Data size and memory. Does the dataset fit in RAM, and what happens when it does not? Pandas materializes everything eagerly and dies when it overflows; the senior answer names out-of-core streaming — an engine that processes data in bounded batches and spills to disk — as the property that separates a toy from a tool.
Data modality. Is the data purely tabular, or does it include images, audio, tensors, and embeddings? This is the axis that eliminates most tools: a DataFrame that cannot hold a decoded image as a typed column forces you to shuttle file paths around and do the real work outside the engine. The senior answer names multimodal columns as a first-class requirement, not an afterthought.
Execution model. Eager (every line runs immediately) or lazy (transformations build a plan a query optimizer can rewrite)? Eager is easy to debug but forfeits optimization; lazy lets the engine push filters into the scan and read only the columns it needs. The senior answer prefers lazy plus an optimizer for anything that touches real data volumes.
Infrastructure. One machine or a cluster — and does scaling out mean rewriting the program? Tools that make you port a script to a distributed dialect impose a tax on every experiment. The senior answer values the same code, two runners — a local engine and a distributed one behind one API.

What Daft actually is — the shape of the tool.

A Rust engine with a Python API. The DataFrame you write is Python; the execution — I/O, decoding, aggregation, joins — runs in a native Rust engine over Arrow buffers, so there is no per-row Python interpreter tax and no JVM.
Lazy and optimized. Operations build a logical plan. Nothing reads data until you call an action (.collect(), .show(), .write_parquet()), at which point Daft's optimizer rewrites the plan — pushing predicates and projections into the scan — and the engine runs it.
Columnar and typed. Data lives in Apache Arrow columnar format with a real type system that includes image, tensor, and embedding alongside the usual numerics and strings — the representation that makes multimodal work first-class.
Local or distributed on Ray. The default native runner is a multithreaded, streaming, out-of-core engine for a single machine; flip the runner and the identical program executes distributed across a Ray cluster.

What interviewers listen for.

Do you name out-of-core / streaming execution as the answer to "the data doesn't fit in memory," rather than "use a bigger box"? — senior signal.
Do you treat multimodal columns (images, tensors, embeddings) as a typed first-class concern, not "store the S3 path and decode later"? — senior signal.
Do you prefer lazy evaluation with a query optimizer and can you explain predicate/projection pushdown? — required answer.
Do you value one code path across a local and a distributed runner over rewriting for the cluster? — senior signal.
Do you pick the engine by the workload (size, modality, infra) rather than by habit? — required answer.

Worked example — the DataFrame-engine decision table

Detailed explanation. The single most useful artifact for an engine-choice interview is a memorised mapping of workload → engine. Every senior discussion converges on it: given the data size, the modality, and the infrastructure, do you reach for Pandas, Polars, Spark, or Daft? Walk through building the table for a team that starts with a 2 GB CSV and grows into a 50 TB image corpus.

The workloads. A small tabular file on a laptop; a large tabular job on a cluster; a multimodal (image + embedding) job that must scale.
The tension. Familiarity and single-node simplicity pull toward Pandas/Polars; scale pulls toward Spark; modality pulls toward Daft — and only Daft covers scale and modality without a JVM.
The rule. Choose by the axis that eliminates the most tools first: modality and out-of-core needs usually decide it before size does.

Question. For each workload, name the engine and the property that makes it the right pick.

Input.

Workload	Size	Modality	Best pick
Laptop tabular exploration	fits in RAM	tabular	Pandas / Polars
Cluster tabular ETL	> RAM, petabyte-scale	tabular	Spark or Daft
Larger-than-memory on one box	> RAM, one machine	tabular	Daft (out-of-core)
Images + embeddings at scale	> RAM, distributed	multimodal	Daft

Code.

import daft
from daft import col

# The SAME Daft program covers the last three rows of the table.
# Lazy: this builds a plan, it does not read the 50 TB yet.
df = daft.read_parquet("s3://corpus/images/*.parquet")   # tabular + a URL column
df = df.where(col("split") == "train")                    # predicate -> pushed into scan
df = df.select(col("image_url"), col("label"))            # projection -> only 2 columns read

# Multimodal, first-class: a decoded image is a typed column, not a path on the side.
df = df.with_column("image", col("image_url").url.download().image.decode())

# Trigger execution — locally today, on a Ray cluster tomorrow, unchanged.
df.show(3)

Step-by-step explanation.

daft.read_parquet(...) does not read the corpus; it registers a scan in a lazy plan. That laziness is what later lets the optimizer read only the split=='train' row groups and only the two selected columns — the difference between scanning 50 TB and scanning a sliver of it.
The where and select are plan nodes, not immediate work. Pandas would have materialized an intermediate frame at each step; Daft records the intent and defers, so it can fold the filter and the column list back into the Parquet reader.
col("image_url").url.download().image.decode() is the axis that eliminates Spark from this row: the decoded image is a genuine typed column the engine understands, chainable like any expression — no shuttling paths to a separate decode step outside the DataFrame.
df.show(3) is the action that triggers everything: the optimizer runs, then the Rust engine executes. Because Daft is out-of-core, this works whether the input is 2 GB or 50 TB — it processes in bounded batches rather than loading it all.
The mistake the table prevents is picking by habit: reaching for Pandas on the 50 TB image job (out of memory), or for Spark on the multimodal job (no image column). Modality and out-of-core needs decide it before raw size does.

Output.

Deciding axis	Eliminates	Leaves
Larger-than-memory	Pandas (eager, in-RAM)	Polars (streaming), Spark, Daft
Multimodal columns	Spark, most SQL engines	Daft
No JVM / Python-first	Spark	Daft, Polars
Scale + modality together	Polars (single node)	Daft

Rule of thumb. Choose a DataFrame engine by the axis that eliminates the most tools first — usually modality (can it hold an image/embedding?) and out-of-core (does it survive larger-than-memory?). For tabular data that fits in RAM, Pandas or Polars is fine; the moment the workload is big and multimodal, Daft is the pick that covers both without a JVM.

Worked example — what interviewers actually probe

Detailed explanation. The senior "why this tool" interview escalates predictably: an innocent opener ("just use Pandas?"), then narrowing pressure to test whether you understand memory, modality, laziness, and distribution. Candidates who name out-of-core streaming, multimodal columns, and the local-to-Ray path score highest.

Ambiguous opener. "It's a DataFrame job — Pandas, right?"
Follow-up 1. "The input is 3 TB and won't fit in memory." — probes out-of-core.
Follow-up 2. "Each row has an image we must decode and embed." — probes multimodal.
Follow-up 3. "It's too slow reading everything." — probes lazy + pushdown.
Follow-up 4. "Now run it on the cluster." — probes distribution without a rewrite.

Question. Draft a crisp senior answer that pre-empts all four follow-ups without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Big data	"sample it down"	"out-of-core engine streams + spills"
Multimodal	"store S3 paths, decode later"	"images/embeddings as typed columns"
Slow reads	"read faster hardware"	"lazy plan; push filters/columns into the scan"
Cluster	"rewrite in Spark"	"same Daft code, switch to the Ray runner"
Framing	"pick what I know"	"pick by size, modality, and infra"

Code.

Senior "why Daft" answer template
=================================

1 — name the two failure modes up front
  "Pandas dies on data bigger than memory; Spark can't hold an image or
   an embedding as a column. Daft is a Rust-engine DataFrame that is
   out-of-core AND multimodal-native, so it covers both."

2 — memory
  "It streams in bounded batches and spills to disk, so a 3 TB input
   runs on a machine with far less RAM — no down-sampling."

3 — modality
  "URL, image, tensor, and embedding are first-class typed columns.
   download -> decode -> embed is a few expressions and a UDF, inside
   the DataFrame, not glue bolted on outside it."

4 — speed / laziness
  "It's lazy: transformations build a plan the optimizer rewrites,
   pushing predicates and projections into the Parquet scan so it reads
   only the rows and columns the query needs."

5 — scale
  "The same program runs on the native runner locally or on a Ray
   cluster by switching the runner — I don't rewrite it to scale out."

Step-by-step explanation.

Point 1 frames the whole answer around the two failure modes — memory and modality — because naming both is what signals you understand where Pandas and Spark each break, not just that Daft is "another DataFrame."
Point 2 answers the memory follow-up with out-of-core streaming, the property that lets a big input run on a small machine; "down-sample" is the tell of someone who has never processed data that genuinely did not fit.
Point 3 answers the multimodal follow-up by making media typed columns — the difference between a DataFrame that understands an image and one that just carries a string path you decode elsewhere.
Point 4 answers the speed follow-up with laziness and pushdown, showing you know why a lazy engine reads less: the optimizer folds filters and column lists into the scan before any bytes move.
Point 5 closes on distribution without a rewrite — the single most senior thing you can say about an engine choice, because it means every experiment you ran locally is already a cluster job.

Output.

Grading criterion	Weak score	Senior score
Names memory + modality failure modes	rare	mandatory
Out-of-core over down-sampling	occasional	senior signal
Multimodal as typed columns	rare	senior signal
Lazy + pushdown for speed	occasional	mandatory
Local-to-Ray without a rewrite	rare	senior signal

Rule of thumb. The senior "why this engine" answer names the two failure modes it fixes — bigger-than-memory and multimodal — then covers out-of-core streaming, typed media columns, lazy pushdown, and one-code-path distribution, without waiting for the follow-ups. Rehearse it once; it reframes an engine question into an architecture answer.

Worked example — Daft vs Spark vs Ray Data vs Polars

Detailed explanation. A common trap is "isn't this just Spark / Polars / Ray Data?" The weak answer picks by popularity. The senior answer places each tool on the size/modality/model axes and shows precisely where Daft overlaps and where it is distinct. Walk the four for the same image-embedding job.

Polars. A superb single-machine Rust DataFrame with a streaming engine — but single-node and tabular-first, so it stops at the cluster and at multimodal.
Spark. The distributed workhorse for tabular ETL — but JVM-based, with a Python serialization boundary and no native image/embedding column.
Ray Data. A distributed dataset for ML preprocessing on Ray — great at streaming blocks into training, but it is a lower-level dataset, not a full DataFrame with a SQL-grade query optimizer.
Daft. A Python-first, Rust-engine DataFrame that is lazy and optimized, multimodal-native, and runs local or on Ray — the one that covers distributed and multimodal and a query optimizer at once.

Question. Contrast the four on execution engine, distribution, multimodal support, and query optimization for a scaled image-embedding pipeline.

Input.

Dimension	Polars	Spark	Ray Data	Daft
Engine	Rust, single node	JVM	Python/Ray	Rust
Distribution	no (one machine)	yes	yes (Ray)	yes (Ray)
Multimodal columns	limited	no	via Python objects	first-class typed
Query optimizer	yes (streaming)	yes (Catalyst)	minimal	yes
Python-native	yes	boundary	yes	yes

Code.

import daft
from daft import col, DataType

# Daft: one program that is distributed AND multimodal AND optimized.
df = daft.read_parquet("s3://corpus/*.parquet")            # lazy, columnar
df = df.with_column("img", col("url").url.download().image.decode())
df = df.with_column("thumb", col("img").image.resize(224, 224))

# A GPU batch UDF the DataFrame runs distributed on Ray — Spark can't type this column,
# Polars can't distribute it, Ray Data would make you hand-roll the query logic.
@daft.udf(return_dtype=DataType.embedding(DataType.float32(), 512), num_gpus=1, concurrency=4)
class Embed:
    def __init__(self):
        self.model = load_encoder().cuda().eval()
    def __call__(self, images):
        return self.model.encode(images.to_pylist())

df = df.with_column("embedding", Embed(col("thumb")))
df.write_parquet("s3://corpus/embeddings/")                # action: optimize + run

Step-by-step explanation.

The read and the two with_column image steps are the part Spark cannot express: image.decode() and image.resize(...) operate on a typed image column, so decoding is inside the engine and the optimizer can still reason about the plan around it.
The Embed UDF is declared with num_gpus=1 and concurrency=4, so Daft schedules it across GPU workers — this is where Ray Data would also do well, but you would be writing the read, filter, and column logic yourself instead of getting a DataFrame with an optimizer for free.
Polars is eliminated at the write_parquet scale point: it is a single-machine engine, so a 50 TB corpus with GPU embedding across a cluster is out of its scope, however fast it is on one box.
Every distributed engine here (Spark, Ray Data, Daft) will parallelize the work; the distinguishing question is whether the tool gives you both a query optimizer and first-class multimodal columns — and only Daft answers yes to both.
The senior move is not "Daft is best" but "Daft is the intersection": pick Polars for fast single-node tabular, Spark for JVM-shop tabular ETL, Ray Data for a pure ML block stream — and Daft when you need distributed, multimodal, and a real query plan together.

Output.

Need	Reach for
Fast, single-machine tabular	Polars
Petabyte tabular ETL in a JVM shop	Spark
Pure ML block stream into training	Ray Data
Distributed + multimodal + optimizer	Daft

Rule of thumb. Place the tools on axes, not a leaderboard: Polars owns single-node tabular, Spark owns JVM tabular ETL, Ray Data owns the ML block stream — and Daft owns the intersection where you need distribution, first-class multimodal columns, and a query optimizer at the same time. It is a per-workload decision, not a rivalry.

Senior interview question on choosing a DataFrame engine for multimodal ML

A senior interviewer often opens with: "Your team has a 40 TB corpus of product images in S3 with a metadata table, and you need to filter to a training split, decode and resize each image, compute an embedding on the GPU, and write the vectors back — today on one big machine, next quarter on a cluster. Pandas runs out of memory, Spark can't hold the image column, and a hand-written Ray script is becoming a maintenance burden. Choose an engine and justify it on memory, modality, execution model, and the path from laptop to cluster."

Solution Using a lazy, columnar, multimodal Daft DataFrame that scales from native to Ray

import daft
from daft import col, DataType

# 1 — Lazy, columnar read. Predicate + projection are pushed into the Parquet scan,
#     so only the training split and the needed columns are ever read from S3.
df = daft.read_parquet("s3://corpus/products/*.parquet")
df = df.where(col("split") == "train").select(col("image_url"), col("sku"))

# 2 — Multimodal columns: URL -> bytes -> image -> resized image, all typed, in-engine.
df = df.with_column("bytes", col("image_url").url.download(on_error="null"))
df = df.with_column("image", col("bytes").image.decode())
df = df.with_column("thumb", col("image").image.resize(224, 224))

# 3 — GPU batch inference as a class UDF: model loaded once per worker, images batched.
@daft.udf(return_dtype=DataType.embedding(DataType.float32(), 512),
          num_gpus=1, concurrency=8, batch_size=64)
class Embed:
    def __init__(self):
        self.model = load_encoder().cuda().eval()
    def __call__(self, images):
        return self.model.encode(images.to_pylist())

df = df.with_column("embedding", Embed(col("thumb")))
df = df.exclude("bytes", "image")            # drop heavy intermediate columns before write
df.write_parquet("s3://corpus/embeddings/")  # ACTION: optimize the plan, then run

# 4 — Same program, two runners. Nothing above changes; only the runner does.
import daft
daft.context.set_runner_native()                       # local: multithreaded, out-of-core
# daft.context.set_runner_ray(address="ray://head:10001")  # cluster: distributed on Ray

Step-by-step trace.

Decision	Before (Pandas / Spark / ad-hoc Ray)	After (Daft)
Bigger-than-memory input	Pandas OOM; sample down	streamed out-of-core, no sampling
Image column	Spark: path string only	typed `image` column, decoded in-engine
Reading 40 TB	full scan	pushdown: split + 2 columns only
GPU embedding	bespoke Ray glue	one class UDF, `num_gpus`/`concurrency`
Laptop → cluster	rewrite in Spark	switch runner, code unchanged
Query optimization	manual in Ray	optimizer folds filter/projection into scan

After the rollout, the pipeline reads only the training split and two columns from S3 (predicate and projection pushdown), downloads and decodes each image into a typed column, resizes it, and runs a GPU-batched embedding UDF that loads the model once per worker — all as one lazy plan the optimizer rewrites before the Rust engine streams it. The identical program runs multithreaded and out-of-core on the single big machine today and distributed on Ray next quarter, with only the runner line changing. The 40 TB never lands in memory at once and no image work happens outside the DataFrame.

Output:

Metric	Pandas / Spark / ad-hoc	Daft
Peak memory vs data size	must fit (or sample)	bounded (streams + spills)
Image/embedding column	external glue	first-class typed column
Bytes read from S3	whole files	pushed-down slice only
Code to scale to a cluster	a rewrite	one runner switch
Query optimization	hand-written	automatic (optimizer)

Why this works — concept by concept:

Lazy plan with pushdown — transformations build a plan the optimizer rewrites, folding the where and select into the Parquet scan so only the training split and two columns are read, not the whole 40 TB. Reading less is the cheapest optimization there is.
Multimodal typed columns — URL, bytes, image, and embedding are real column types the Rust engine understands, so decode-resize-embed happens inside the DataFrame and the optimizer can still reason about the surrounding plan.
Rust engine, out-of-core — the native engine streams data in bounded batches and spills to disk, so a 40 TB input runs on a machine that could never hold it, with no per-row Python tax on the heavy work.
One code path, two runners — the same program runs on the native runner locally and on Ray distributed, so scaling out is a configuration change, not a rewrite, and every local experiment is already a cluster job.
Cost — one model load per worker, a pushed-down scan, and streamed batches, versus a full scan plus a bespoke distributed script plus per-row Python. The eliminated cost is an entire hand-rolled query engine and its rewrite-for-scale tax — O(read what you need) instead of O(read and materialize everything).

Design
Topic — design
Design problems on data-engine and pipeline architecture

Practice →

Data processing Topic — data-processing Data processing problems on DataFrames and large datasets

Practice →

2. The execution model — lazy DataFrame, optimizer, columnar

Transformations build a plan; the optimizer rewrites it; the Rust engine runs it only when you collect

The mental model in one line: a Daft DataFrame is lazy — every where, select, with_column, join, and groupby appends a node to a logical plan instead of touching data — and only an action (.collect(), .show(), .write_parquet(), .to_pandas()) triggers execution, at which point the query optimizer rewrites the plan (predicate pushdown, projection pushdown, limit pushdown, filter reordering) and the Rust engine runs it over Apache Arrow columnar buffers, so the difference between a slow job and a fast one is usually how much the optimizer could prune before a single byte moved. You describe what result you want; Daft decides how to compute it with the least I/O and memory.

Lazy by default — plans, not results.

Transformations are plan nodes. read_*, where, select, with_column, join, groupby/agg, sort, limit all return a new DataFrame that records the operation; none of them read or compute data.
Actions trigger execution. .collect() materializes into memory, .show(n) previews, .write_parquet(...) streams to storage, .to_pandas()/.to_arrow() hand off — each is the point where the optimizer runs and the engine executes.
.explain() shows the plan. df.explain(show_all=True) prints the unoptimized logical plan, the optimized logical plan, and the physical plan — the single most useful debugging habit, because it shows what the optimizer actually pushed down.
Why lazy wins. Eager execution (Pandas) computes each intermediate immediately and cannot see the whole query; a lazy plan lets the optimizer read fewer columns, skip row groups, and reorder work before anything runs.

The query optimizer — do less I/O.

Predicate pushdown. A where filter is pushed into the scan, so a Parquet reader uses row-group statistics to skip blocks and a partitioned dataset skips whole partitions — the query reads only rows that can match.
Projection pushdown. A select of two columns pushes the column list into the scan, so a columnar file reads only those two column chunks instead of every column in the row.
Limit pushdown and early stop. A limit lets the engine stop reading once enough rows are produced, so a preview never scans the whole dataset.
Column pruning through the plan. The optimizer propagates which columns are actually needed downstream, dropping any the final result never uses — so intermediate steps never carry dead columns.

Columnar Arrow memory — why Rust is fast here.

Columnar layout. Values of a column are stored contiguously (Arrow), so vectorized operations run over tight arrays and the CPU cache and SIMD are used well — the representation columnar analytics is built on.
A real type system. Numerics, strings, temporals, nested lists/structs, and the multimodal types (image, tensor, embedding, python) — the engine knows each column's type and picks the right kernel.
No per-row Python. Because the engine is Rust operating on Arrow buffers, filtering, arithmetic, and aggregation never enter the Python interpreter; Python only describes the plan.
Zero-copy hand-offs. Arrow is the interchange format, so .to_arrow() and hand-offs to other Arrow-native tools avoid re-serializing the data.

The failure modes senior engineers pre-empt.

Expecting eager results. Printing a DataFrame variable and being surprised nothing ran — because it is a plan. Mitigation: call an action (.show(), .collect()) and read .explain() to see what will execute.
Defeating pushdown. Wrapping a column in an opaque Python UDF before filtering forces the engine to compute the UDF on rows a later filter would discard. Mitigation: filter and project first, apply expensive UDFs last, and check the optimized plan.
Materializing too early. Calling .collect() mid-pipeline pulls everything into memory and throws away laziness for the rest. Mitigation: keep the plan lazy end-to-end and let the final action stream.

Common interview probes on the execution model.

"Is Daft eager or lazy?" — lazy; transformations build a plan, actions execute it.
"What does the optimizer do?" — predicate, projection, and limit pushdown, plus column pruning, so the scan reads only needed rows and columns.
"Why columnar / Arrow?" — vectorized, cache- and SIMD-friendly, typed, and zero-copy interchange — and it keeps the heavy work in Rust, out of Python.
"How do you debug a slow query?" — df.explain(show_all=True) and confirm filters/columns were pushed into the scan.

Worked example — build a lazy plan and read `.explain()`

Detailed explanation. The first thing to internalize is that a chain of Daft operations does no work until an action, and .explain() reveals the plan the optimizer produced. Build a small pipeline and read its physical plan to prove the pushdowns happened.

The chain. read → where → select → limit.
The claim. No data is read until .show().
The proof. .explain(show_all=True) shows filter and projection folded into the scan.

Question. Construct a lazy pipeline over a Parquet dataset and describe what .explain(show_all=True) reveals about pushdown.

Input.

Step	Call	Plan effect
read	`read_parquet(...)`	scan node (no I/O yet)
filter	`.where(col("country")=="US")`	predicate to push down
project	`.select("user_id","revenue")`	projection to push down
limit	`.limit(100)`	limit to push down
action	`.show(5)`	optimize + execute

Code.

import daft
from daft import col

df = daft.read_parquet("s3://events/2026/*.parquet")   # (1) scan node, nothing reads yet
df = df.where(col("country") == "US")                  # (2) predicate recorded
df = df.select(col("user_id"), col("revenue"))         # (3) projection recorded
df = df.limit(100)                                     # (4) limit recorded

# Inspect the plan BEFORE running: unoptimized -> optimized -> physical.
df.explain(show_all=True)
# Optimized physical scan shows, roughly:
#   ParquetScan
#     Pushdowns: filter=(country == 'US'), columns=[user_id, revenue], limit=100
# i.e. the filter, the 2-column projection, and the limit were folded INTO the read.

df.show(5)   # (5) THE action: optimizer runs, Rust engine executes, 5 rows returned

Step-by-step explanation.

Steps 1–4 return DataFrames but read nothing: each call appends a node (scan, filter, projection, limit) to the logical plan. If you stopped here, zero bytes would move — the plan is inert until an action.
.explain(show_all=True) prints three plans. The unoptimized logical plan lists the operations in the order you wrote them; the optimized one shows the rewrite; the physical one shows exactly what the engine will execute.
In the optimized physical plan, the Parquet scan carries Pushdowns: — the country == 'US' predicate, the [user_id, revenue] column list, and the limit=100 are attached to the scan itself, so the reader skips non-matching row groups and reads only two columns.
.show(5) is the trigger: only now does the optimizer finalize and the Rust engine run. Because the filter and projection are in the scan, the engine reads a fraction of the dataset, not all of it.
The lesson the plan teaches: laziness is not a delay, it is an opportunity — deferring execution is what lets the optimizer see the whole query and read the minimum. Reading .explain() is how you confirm it actually did.

Output.

Plan stage	What it shows
Unoptimized logical	read → filter → project → limit, as written
Optimized logical	filter + projection + limit pushed toward the scan
Physical	`ParquetScan` with `Pushdowns: filter, columns, limit`
Bytes read	only US row groups, only 2 columns, ≤ 100 rows

Rule of thumb. Treat a Daft chain as a plan, not a result: nothing runs until an action, so reach for .explain(show_all=True) to confirm the optimizer pushed your filters and column list into the scan. If the pushdowns are missing, the query is reading more than it needs — and the plan will tell you exactly where.

Worked example — predicate and projection pushdown into a Parquet scan

Detailed explanation. Pushdown is the optimization that matters most on real data, because reading less is faster than processing faster. Contrast a naive full read with a pushed-down read and quantify what each touches.

Naive. Read all columns and all rows, then filter and select in memory.
Pushed down. The optimizer folds the filter and column list into the scan.
The win. I/O drops from "the whole dataset" to "the matching slice."

Question. Show how a where + select on a partitioned, columnar dataset turns into a scan that reads only matching partitions and only the requested columns.

Input.

Aspect	Naive (eager)	Pushed down (Daft)
Rows read	all	only matching partitions/row groups
Columns read	all	only selected
Where it filters	in memory, after read	in the scan
I/O	whole dataset	matching slice

Code.

import daft
from daft import col

# Dataset partitioned by dt=YYYY-MM-DD, columnar Parquet, ~200 columns per row.
df = daft.read_parquet("s3://lake/events/")            # hive-partitioned by `dt`

result = (
    df
    .where(col("dt") == "2026-08-25")                  # partition predicate -> skip other days
    .where(col("event_type") == "purchase")            # row-group predicate -> skip blocks
    .select(col("user_id"), col("amount_cents"))       # projection -> read 2 of ~200 columns
    .collect()
)

# The optimized physical plan attaches these to the scan:
#   ParquetScan(
#     partition_filter = dt = '2026-08-25',            # entire other partitions skipped
#     row_filter       = event_type = 'purchase',      # row groups pruned by statistics
#     columns          = [user_id, amount_cents],      # only 2 column chunks read
#   )

Step-by-step explanation.

The first where on the partition column dt becomes a partition filter: Daft never opens the files for other days at all, so a 365-day lake read collapses to a single day's files before any row is examined.
The second where on event_type becomes a row-group filter: Parquet stores min/max statistics per row group, so blocks that cannot contain purchase are skipped without decoding them.
The select becomes a projection: a columnar file stores each column separately, so reading user_id and amount_cents touches two column chunks and ignores the other ~198 — the single biggest I/O saving on wide tables.
.collect() triggers it, and because all three prunings live in the scan, the engine reads a tiny fraction of the dataset. The naive approach would have read every column of every day and then discarded almost all of it in memory.
The senior insight: pushdown is why lazy beats eager on real data — the optimizer can only prune the scan because it deferred execution long enough to see the whole query. Order does not matter to you (Daft reorders), but expressibility does: keep filters as engine expressions, not opaque UDFs, so they can be pushed.

Output.

Metric	Naive full read	Pushed-down read
Partitions opened	365	1
Row groups decoded	all	only matching
Columns read	~200	2
Data moved	whole lake slice	matching sliver

Rule of thumb. Let the optimizer read less: filter on partition and indexed/statistic-friendly columns and select only the columns you need, and Daft folds all of it into the scan. Keep predicates as native expressions (not opaque Python) so they remain pushable — reading less is the optimization that beats every in-memory speed-up.

Worked example — typed columns and expressions over Arrow

Detailed explanation. Daft's speed and its multimodal power both come from the same place: typed, columnar data with a rich expression API that compiles to Rust kernels. Build a few derived columns and see why none of the work enters Python.

Expressions. col(...), arithmetic, comparisons, string/temporal ops, and .cast(...) build column expressions.
Types. Each column has an Arrow-backed dtype the engine dispatches on.
No Python per row. The kernels run in Rust over contiguous arrays.

Question. Add derived columns with expressions and casts, and explain why the computation is vectorized in Rust rather than looped in Python.

Input.

New column	Expression	Type
`revenue`	`col("amount_cents") / 100`	float64
`is_big`	`col("amount_cents") > 10_000`	bool
`day`	`col("ts").dt.date()`	date
`region_up`	`col("region").str.upper()`	string

Code.

import daft
from daft import col

df = daft.read_parquet("s3://lake/orders/*.parquet")

df = df.select(
    col("order_id"),
    (col("amount_cents") / 100).alias("revenue"),        # arithmetic kernel (Rust)
    (col("amount_cents") > 10_000).alias("is_big"),      # comparison kernel (Rust)
    col("ts").dt.date().alias("day"),                    # temporal accessor
    col("region").str.upper().alias("region_up"),        # string accessor
)

# Each expression is a typed operation the Rust engine runs over Arrow arrays.
# There is NO Python-level per-row loop: Python only built the plan above.
df.show(3)

Step-by-step explanation.

Each col(...) expression describes a typed transformation, and .alias(...) names the output column. Because the input dtypes are known (amount_cents is an integer, ts is a timestamp), Daft dispatches the correct Rust kernel for each.
The arithmetic (/ 100) and comparison (> 10_000) run as vectorized operations over the contiguous Arrow arrays — one tight loop in Rust over the whole column, using SIMD, not a Python for over rows.
The .dt.date() and .str.upper() accessors are typed namespaces: they exist because the engine knows ts is temporal and region is a string, and they too compile to native kernels rather than Python calls.
Nothing here crosses into the Python interpreter at execution time — Python ran once to build the plan, and the engine runs the plan in Rust. This is why Daft avoids the per-row overhead that makes Pandas apply and Spark Python UDFs slow.
The same typed-column machinery is what makes multimodal work first-class: image, tensor, and embedding are just more dtypes with their own accessors (.image.decode(), etc.), covered in section 4 — the expression system is uniform across tabular and media columns.

Output.

Column	Kernel	Runs in
`revenue`	float divide	Rust (vectorized)
`is_big`	int compare	Rust (vectorized)
`day`	temporal extract	Rust
`region_up`	string upper	Rust

Rule of thumb. Express transformations as Daft column expressions and let the typed, Arrow-backed engine run them as vectorized Rust kernels — never drop to a per-row Python loop for something an expression can do. The same typed-column system that makes tabular ops fast is what makes images, tensors, and embeddings first-class.

Senior interview question on Daft's lazy execution and optimizer

A senior interviewer might ask: "Explain Daft's execution model end to end for a query that reads a wide, partitioned Parquet dataset, filters to one partition and one event type, selects two columns, and aggregates. Cover what is lazy versus eager, what the optimizer does before execution, why the columnar Arrow representation matters, and how you would prove — not assume — that the filters and projection were pushed into the scan."

Solution Using a lazy plan, pushdown optimization, columnar execution, and `.explain()`

import daft
from daft import col

# 1 — Build the plan lazily. None of these lines read data.
df = daft.read_parquet("s3://lake/events/")            # hive-partitioned by dt
plan = (
    df
    .where(col("dt") == "2026-08-25")                  # partition predicate
    .where(col("event_type") == "purchase")            # row-group predicate
    .select(col("region"), col("amount_cents"))        # projection (2 of ~200 cols)
    .groupby(col("region"))
    .agg(col("amount_cents").sum().alias("revenue_cents"))
)

# 2 — PROVE the optimization before running a single byte of I/O.
plan.explain(show_all=True)
# Physical plan (abridged):
#   ParquetScan
#     partition_filter = (dt == '2026-08-25')          # other partitions never opened
#     row_filter       = (event_type == 'purchase')    # row groups pruned by statistics
#     columns          = [region, amount_cents]        # only 2 column chunks read
#   -> Aggregate(group=[region], sum(amount_cents))     # vectorized in Rust over Arrow

# 3 — Trigger execution. Optimizer finalizes, Rust engine streams the aggregate.
result = plan.collect()

# 4 — Anti-pattern to AVOID: an opaque UDF before the filter defeats pushdown.
#   df.with_column("x", slow_udf(col("payload"))).where(col("dt") == "2026-08-25")
#   -> the UDF may run on rows the filter would have discarded. Filter FIRST, UDF LAST.

Step-by-step trace.

Layer	Component	Purpose
Plan	lazy DataFrame	records ops, reads nothing
Optimize	predicate pushdown	skip non-matching partitions/row groups
Optimize	projection pushdown	read 2 columns, not ~200
Execute	Rust engine over Arrow	vectorized filter + aggregate
Prove	`.explain(show_all=True)`	pushdowns visible on the scan
Trigger	`.collect()`	run only at the action

After building the plan, .explain(show_all=True) shows the partition filter, the row-group filter, and the two-column projection all attached to the Parquet scan, with the group-by aggregation above it — proof, not assumption, that the optimizer pruned the read. .collect() then runs the optimized physical plan: the engine opens only the 2026-08-25 partition, skips row groups that cannot contain purchases, reads only region and amount_cents, and aggregates them vectorized in Rust. The wide, partitioned dataset is touched only where the query can match, and no per-row Python runs.

Output:

Metric	Eager (Pandas-style)	Daft lazy + optimized
Partitions read	all	one
Columns read	~200	2
Filter location	in memory after read	in the scan
Aggregation	Python loop / boxed	vectorized Rust
Verifiable plan	none	`.explain()` shows pushdowns

Why this works — concept by concept:

Lazy plan — every transformation appends to a logical plan and reads nothing, so the optimizer gets to see the whole query at once and prune it before any I/O — laziness is the precondition for every pushdown that follows.
Predicate and projection pushdown — the filters fold into the scan as partition and row-group pruning and the select folds in as a column list, so the reader touches only the matching partition, the matching row groups, and two columns instead of the entire wide dataset.
Columnar Arrow execution — contiguous typed arrays let the Rust engine filter and aggregate with vectorized, cache- and SIMD-friendly kernels, with no per-row Python interpreter tax on the hot path.
.explain() as proof — printing the optimized physical plan turns "I think it pushed down" into "the scan carries these pushdowns," the discipline that catches an opaque UDF silently defeating the optimizer.
Cost — reading one partition and two columns and aggregating in Rust, versus scanning ~200 columns of every partition and looping in Python. The eliminated cost is the I/O and memory of everything the query never needed — O(matching slice) instead of O(whole dataset).

Optimization
Topic — optimization
Optimization problems on query pushdown and lazy evaluation

Practice →

Data processing Topic — data-processing Data processing problems on columnar transforms and aggregation

Practice →

3. Distributed execution on Ray — local to cluster

The same lazy plan runs multithreaded on one machine or distributed across a Ray cluster

The mental model in one line: Daft has one program and two engines — the native runner (a multithreaded, streaming, out-of-core Rust engine for a single machine) and the Ray runner (which distributes the same logical plan across a cluster) — and you choose between them by setting a runner, not by rewriting code, because Daft partitions the data and turns partition-level work into scheduled tasks that run wherever there is capacity, with streaming (morsel-driven) execution and spill-to-disk keeping memory bounded whether you are on a laptop or on a hundred nodes. Develop and test locally on real logic and real (sampled) data; scale to the cluster by flipping the runner and, when needed, tuning partitioning.

One program, two runners.

Native runner (default). A single-node, multithreaded Rust engine with a streaming, morsel-driven execution model — it processes data in small batches through a pipeline, using all cores and spilling to disk when memory is tight, so it handles larger-than-memory inputs on one machine.
Ray runner. The identical plan executed on a Ray cluster: Daft schedules partition-level tasks across workers, moving only the data a step needs, so the same DataFrame program spans many machines and GPUs.
Switching runners. daft.context.set_runner_native() or daft.context.set_runner_ray(address=...) (or the DAFT_RUNNER environment variable) — no operator or transformation in your pipeline changes.
Why this matters. The expensive part of distributed data work is usually the rewrite between local and cluster dialects; Daft removes it, so a laptop notebook and a production cluster job are the same code.

Streaming and out-of-core execution.

Morsel-driven pipeline. Instead of materializing each operator's full output, the native engine pushes small batches (morsels) through the operator pipeline, so peak memory is bounded by the in-flight batches, not the dataset size.
Spill to disk. When an operation (a large sort or aggregation) needs more memory than available, the engine spills intermediate state to disk rather than failing — the mechanism behind "larger-than-memory."
Backpressure. The pipeline slows upstream reads when downstream is busy, so a fast scan does not overwhelm a slow GPU UDF — steady, bounded throughput instead of boom-and-OOM.
Same guarantees distributed. On Ray, streaming happens within tasks and across the shuffle, so the out-of-core property holds at cluster scale.

Partitioning and the shuffle.

Partitions are the unit of parallelism. A distributed DataFrame is a set of partitions; each partition is processed by a task, so more partitions means more parallelism (up to the cluster's cores).
into_partitions / repartition. into_partitions(n) sets the count without a full shuffle where possible; repartition(n, col) hash-partitions by a key so all rows for a key land together — required before a key-local group-by or join.
Shuffles are the expensive step. A repartition, groupby, or join on a key moves data across the network; minimizing and co-locating shuffles is the core distributed-tuning skill.
Reading sets initial partitions. File/row-group layout gives the initial partitioning; a well-partitioned source (many balanced files) parallelizes cleanly, a few huge files do not.

The failure modes senior engineers pre-empt.

Too few or too many partitions. One giant partition serializes the job on one core; a million tiny partitions drown in scheduling overhead. Mitigation: target partitions sized in the low hundreds of MB and roughly a small multiple of cluster cores.
Skew. A hot key (one region, one user) makes one partition huge and one task the straggler. Mitigation: salt the key, pre-aggregate, or repartition on a higher-cardinality key.
Accidental local execution. Forgetting to set the Ray runner and quietly running a cluster-sized job on one node. Mitigation: set the runner explicitly (or via DAFT_RUNNER) and confirm it in logs.

Common interview probes on distribution.

"How do you scale a Daft job from a laptop to a cluster?" — switch the runner to Ray; the code is unchanged.
"How does it handle data bigger than memory?" — streaming morsel-driven execution plus spill-to-disk, on both runners.
"What's the expensive operation in a distributed DataFrame?" — the shuffle (repartition/join/group-by on a key); minimize and co-locate it.
"How do you fix a straggler task?" — it is usually skew; salt or pre-aggregate the hot key, or repartition.

Worked example — same code, native runner to Ray runner

Detailed explanation. The headline property is that scaling out is a configuration change. Take a pipeline, run it on the native runner, then point it at a Ray cluster with no change to the DataFrame logic.

The pipeline. read → filter → group-by → write, identical in both runs.
Local. set_runner_native() — multithreaded, out-of-core, one machine.
Cluster. set_runner_ray(address=...) — distributed across workers.

Question. Show one Daft pipeline executing on the native runner and then on a Ray cluster, with only the runner selection differing.

Input.

Aspect	Native runner	Ray runner
Machines	one	many
Parallelism	threads/cores	tasks across workers
Memory model	streaming + spill	streaming + spill per task
Code changes	none	none (just the runner)

Code.

# pipeline.py — the DataFrame logic is identical for both runners.
import daft
from daft import col

def build():
    df = daft.read_parquet("s3://lake/events/*.parquet")
    df = df.where(col("event_type") == "purchase")
    df = (df.groupby(col("region"))
            .agg(col("amount_cents").sum().alias("revenue_cents")))
    return df

# run_local.py — develop and test on one machine, out-of-core.
import daft
from pipeline import build
daft.context.set_runner_native()            # multithreaded Rust engine, streaming
build().write_parquet("s3://out/revenue_by_region/")

# run_cluster.py — SAME build(), now distributed. Nothing in pipeline.py changed.
import daft
from pipeline import build
daft.context.set_runner_ray(address="ray://head.cluster:10001")   # distribute on Ray
build().write_parquet("s3://out/revenue_by_region/")
# Or set once via environment, no code at all:  export DAFT_RUNNER=ray

Step-by-step explanation.

build() in pipeline.py is the entire data logic — read, filter, group-by, aggregate — and it contains no mention of a runner, threads, or a cluster. It describes the plan; it does not choose how to execute it.
run_local.py selects the native runner, so build().write_parquet(...) runs multithreaded on one machine with streaming, out-of-core execution — exactly what you want for development and for jobs that fit one (large) node.
run_cluster.py imports the same build() and selects the Ray runner with a cluster address. Daft now partitions the scan, schedules the filter and the group-by shuffle as tasks across workers, and writes the output — the identical plan, distributed.
The DAFT_RUNNER=ray environment variable does the same with zero code change, which is how the same script runs locally in CI and distributed in production without a code path per environment.
The senior point: because the logic is runner-agnostic, every local test exercises the real pipeline, and "make it distributed" is an operational decision, not an engineering project — the class of bug where the local and cluster versions drift apart simply does not exist.

Output.

Run	Runner	Executes as	Code delta
local dev	native	multithreaded, one node	—
CI	native (or Ray)	same plan	env var only
production	Ray	tasks across workers	runner line / env
correctness	identical logic	same result	none

Rule of thumb. Keep the DataFrame logic runner-agnostic and choose the engine at the edges — set_runner_native() for one machine, set_runner_ray(address=...) (or DAFT_RUNNER=ray) for the cluster. Scaling out becomes a configuration change, so your local tests and your production job are the same code and can never silently diverge.

Worked example — streaming a larger-than-memory scan out-of-core

Detailed explanation. The property that lets a laptop process data bigger than its RAM is streaming, morsel-driven execution with spill. Walk a job whose input dwarfs memory and see why it does not OOM.

The input. 500 GB of Parquet on a 32 GB machine.
The mechanism. Small batches flow through the pipeline; state spills to disk.
The result. Bounded peak memory, no out-of-memory failure.

Question. Explain how Daft processes a 500 GB scan-filter-aggregate on a 32 GB machine without running out of memory.

Input.

Aspect	Eager (Pandas)	Daft native (streaming)
Load model	whole dataset in RAM	bounded batches
500 GB on 32 GB RAM	OOM crash	runs (spills)
Peak memory	≈ dataset size	≈ in-flight batches
Big aggregation state	in RAM	spills to disk

Code.

import daft
from daft import col

daft.context.set_runner_native()     # streaming, morsel-driven, out-of-core

# 500 GB input on a 32 GB box. This does NOT load 500 GB into memory.
df = daft.read_parquet("s3://lake/huge/*.parquet")           # streamed in batches
df = df.where(col("status") == "active")                     # filter per batch
df = (df.groupby(col("customer_id"))                         # aggregation state spills if large
        .agg(col("amount_cents").sum().alias("total_cents")))

# Streaming write: results are produced and written incrementally, not buffered whole.
df.write_parquet("s3://out/customer_totals/")
# Peak memory ~ a handful of in-flight morsels + spill files, NOT ~500 GB.

Step-by-step explanation.

read_parquet does not slurp 500 GB; the native engine pulls the scan in small batches (morsels) and pushes each through the operator pipeline, so at any instant only a few batches are resident.
The where filter runs per morsel as it flows past, so filtered-out rows are dropped early and never accumulate — the pipeline is a river, not a reservoir.
The groupby/agg maintains aggregation state (a running sum per customer_id). If that state grows beyond memory, the engine spills partial state to disk and merges it back, which is what makes an aggregation over a huge, high-cardinality key survive on a small box.
write_parquet streams output as it is produced rather than materializing the full result first, so the write side is bounded too — the job never needs the whole input or the whole output in memory.
Peak memory is therefore a function of the in-flight batch size and spill configuration, not the dataset size — the exact property that separates an out-of-core engine from Pandas, which must fit everything in RAM and crashes at row one of "too big."

Output.

Stage	Memory behavior
scan	a few morsels resident
filter	drops rows per batch
aggregate	spills state to disk if large
write	streamed incrementally
peak	≈ batches + spill, not 500 GB

Rule of thumb. Trust the streaming engine for larger-than-memory work: Daft processes in bounded morsels and spills big aggregation/sort state to disk, so peak memory tracks the in-flight batch size, not the input size. You do not down-sample to fit — you let the pipeline stream.

Worked example — partitioning for a distributed group-by and join

Detailed explanation. On a cluster, the expensive operation is the shuffle, and partitioning controls it. Tune the partition count and key so a group-by and a join parallelize without skew.

The parallelism unit. Partitions → tasks; too few serialize, too many thrash.
The key. Hash-partition by the join/group key so matching rows co-locate.
The hazard. A hot key skews one partition into a straggler.

Question. Choose a partitioning strategy for a distributed join of a large fact table to a dimension, plus a group-by, avoiding skew and shuffle waste.

Input.

Concern	Bad choice	Good choice
Partition count	1 (serial) or 1e6 (overhead)	~2–4× cluster cores
Join key partitioning	mismatched	both hashed on the key
Hot key	one giant partition	salt / pre-aggregate
Partition size	GBs or KBs	low hundreds of MB

Code.

import daft
from daft import col

daft.context.set_runner_ray(address="ray://head:10001")

fact = daft.read_parquet("s3://lake/orders/*.parquet")       # large
dim  = daft.read_parquet("s3://lake/customers/*.parquet")    # small-ish

# Co-locate join keys: hash-partition both sides on customer_id so matching
# rows land in the same partition -> a shuffle that the join can do locally.
fact = fact.repartition(256, col("customer_id"))             # ~2-4x cores, balanced
dim  = dim.repartition(256, col("customer_id"))

joined = fact.join(dim, on="customer_id")                    # key-local, minimal reshuffle

# Group-by on the same key reuses the partitioning -> no extra shuffle.
result = (joined.groupby(col("customer_id"))
                .agg(col("amount_cents").sum().alias("ltv_cents")))

# Skew guard: if one customer_id is a hot straggler, salt then re-aggregate.
# salted = joined.with_column("k", col("customer_id") + "-" + (col("rand") % 8).cast(str))
result.write_parquet("s3://out/customer_ltv/")

Step-by-step explanation.

repartition(256, col("customer_id")) on both sides hash-partitions them on the same key, so every customer_id lands in the same partition index on the fact and the dimension — the join then matches rows within a partition instead of shuffling the whole fact table against the whole dimension.
The partition count (256) is chosen as a small multiple of cluster cores: enough partitions that every core has work, few enough that scheduling overhead stays negligible and each partition is a healthy low-hundreds-of-MB chunk.
The groupby(customer_id) reuses the existing partitioning, so it needs no additional shuffle — the rows for each key are already co-located from the join. Aligning the join key and the group key is a deliberate shuffle-saving choice.
Skew is the residual risk: if one customer_id is enormous, its partition becomes a straggler that holds up the whole stage. The salting comment sketches the fix — split the hot key into sub-keys, aggregate, then combine — trading a bit of extra work for balanced tasks.
The senior framing: distributed performance is mostly shuffle management. Partition on the key you will join and group by, size partitions sensibly, and watch for skew — get those right and the cluster scales linearly; get them wrong and you have a serial job wearing a cluster's clothes.

Output.

Choice	Effect
256 partitions ≈ cores	balanced parallelism
both sides hashed on key	key-local join, minimal shuffle
group-by reuses partitioning	no extra shuffle
salt the hot key	straggler removed

Rule of thumb. Treat the shuffle as the cost center: hash-partition both join sides on the join key, size partitions to the low hundreds of MB and a small multiple of cluster cores, and align the group-by key with the join key to reuse the partitioning. When one task straggles, suspect skew and salt or pre-aggregate the hot key.

Senior interview question on scaling a Daft pipeline onto Ray

A senior interviewer might ask: "You have a Daft pipeline that reads a 20 TB fact table, joins it to a dimension, aggregates per customer, and it works on a sampled file locally. Take it to a Ray cluster: how you switch runners without touching the logic, how streaming keeps memory bounded, how you partition to make the join and group-by parallelize, and how you diagnose and fix a straggler task caused by a hot key."

Solution Using one runner switch, streaming execution, and key-aligned partitioning

# 1 — Runner-agnostic logic (identical local and distributed).
import daft
from daft import col

def build():
    fact = daft.read_parquet("s3://lake/orders/*.parquet")
    dim  = daft.read_parquet("s3://lake/customers/*.parquet")
    fact = fact.repartition(512, col("customer_id"))       # ~2-4x cluster cores
    dim  = dim.repartition(512, col("customer_id"))         # co-locate join keys
    joined = fact.join(dim, on="customer_id")
    return (joined.groupby(col("customer_id"))
                  .agg(col("amount_cents").sum().alias("ltv_cents")))

# 2 — Scale out by switching the runner. No change to build().
import daft
daft.context.set_runner_ray(address="ray://head.cluster:10001")   # or DAFT_RUNNER=ray
build().write_parquet("s3://out/customer_ltv/")                    # streamed, out-of-core

# 3 — Diagnose + fix a straggler (skewed hot key). Salt, aggregate, re-combine.
import daft
from daft import col

joined = build_join()                                    # up to the join
salted = joined.with_column("bucket", (col("rand") % 16).cast(daft.DataType.int64()))
partial = (salted.groupby(col("customer_id"), col("bucket"))
                 .agg(col("amount_cents").sum().alias("part_cents")))
final = (partial.groupby(col("customer_id"))
                .agg(col("part_cents").sum().alias("ltv_cents")))   # combine buckets
final.write_parquet("s3://out/customer_ltv/")

Step-by-step trace.

Layer	Component	Purpose
Logic	runner-agnostic `build()`	same code local + cluster
Scale	`set_runner_ray()`	distribute the plan on Ray
Memory	streaming + spill	20 TB on bounded RAM per worker
Parallelism	`repartition(512, key)`	balanced tasks, key co-located
Join + group	key-aligned partitioning	one shuffle serves both
Skew fix	salt → aggregate → combine	straggler removed

After scaling, the identical build() runs on Ray: the 20 TB scan streams in partitions, both tables are hash-partitioned on customer_id so the join is key-local, and the group-by reuses that partitioning for a single shuffle rather than two. Memory stays bounded because each task streams morsels and spills large state. When one enormous customer straggles a task, salting splits that key into sixteen buckets that aggregate in parallel and then re-combine — turning one giant task into sixteen balanced ones. The laptop code and the cluster code are byte-identical; only the runner and the partition count changed.

Output:

Metric	Naive port	Daft on Ray
Code change to distribute	rewrite	one runner line
20 TB memory	must fit / sample	streamed + spilled
Join shuffle	full cross-shuffle	key-local
Group-by shuffle	separate shuffle	reuses join partitioning
Straggler from hot key	stalls the stage	salted into balanced tasks

Why this works — concept by concept:

One runner switch — the DataFrame logic never names a runner, so set_runner_ray() distributes the exact plan that ran locally; there is no local-vs-cluster code drift and every local test exercises the real pipeline.
Streaming, out-of-core execution — morsel-driven pipelines and spill-to-disk keep each worker's memory bounded by in-flight batches, so a 20 TB job runs on workers that could never hold it, on either runner.
Key-aligned partitioning — hash-partitioning both join sides on customer_id co-locates matching rows so the join is local, and aligning the group-by key reuses that partitioning so one shuffle serves both operations.
Skew handling by salting — splitting a hot key into buckets that aggregate in parallel and then combine converts a single straggler task into many balanced ones, the standard cure for the one-hot-key stall.
Cost — one shuffle instead of two, bounded per-worker memory, and balanced tasks, versus a full cross-shuffle, an OOM or a down-sample, and a straggler that serializes the stage. The eliminated cost is the network and time of an unnecessary shuffle plus the straggler tail — O(one key-local shuffle) instead of O(reshuffle-per-operation).

Streaming
Topic — streaming
Streaming problems on out-of-core and morsel-driven execution

Practice →

Design Topic — design Design problems on partitioning, shuffles, and distributed jobs

Practice →

4. Multimodal columns — URLs, images, embeddings

Images, tensors, and embeddings are first-class typed columns — transformed by expressions and UDFs

The mental model in one line: Daft treats multimodal data as first-class typed columns — DataType.image(), DataType.tensor(), DataType.embedding(), and DataType.python() sit alongside numerics and strings — so a URL column becomes bytes via .url.download(), bytes become an image via .image.decode(), an image is resized with .image.resize(), and a model turns images into an embedding column through a UDF, all inside the DataFrame where the optimizer and the distributed engine still apply, instead of the usual pattern of carrying file paths in a tabular frame and doing the real media work in a pile of external Python. The image is data the engine understands, not a string it babysits.

Multimodal types — media as columns.

DataType.image(). A decoded image column (height × width × channels) the engine can resize, crop, and encode with native operations — a real type, not a byte blob you interpret elsewhere.
DataType.tensor() / DataType.embedding(). Fixed- or variable-shape numeric arrays: a tensor for arbitrary N-D data, an embedding(dtype, size) for the fixed-length vectors similarity search and ML consume.
DataType.python(). An escape hatch column holding arbitrary Python objects, for the rare cases a value has no native type — used sparingly because it forfeits vectorization.
Binary and URL as strings. Raw bytes live in a binary column; URLs and paths are ordinary strings until you download them — the pipeline usually starts from a URL/path column.

Expression accessors for media.

.url.download(). Fetches each URL/path (HTTP, S3, local) into a bytes column, with on_error="null" to tolerate dead links instead of failing the job — parallelized by the engine, not a Python loop.
.image.decode(). Turns a bytes column into a typed image column; from there .image.resize(h, w), .image.crop(...), and .image.encode(fmt) are native operations.
Chaining. Because each accessor returns a column expression, col("url").url.download().image.decode().image.resize(224, 224) is one lazy expression the engine fuses.
Still lazy, still optimized. Media expressions live in the same plan as tabular ones, so a filter that drops rows before a download means those images are never fetched.

UDFs over media — batch and class-based.

Batch function UDF. @daft.udf(return_dtype=...) wraps a function that receives a Series (a batch) and returns a batch — vectorized at the Python boundary, so per-call overhead amortizes across many rows.
Class-based UDF for models. A @daft.udf(...) on a class with an initializer and a call method loads the model once per worker (in __init__) and processes batches in the call — the correct pattern for expensive model setup.
Resources and concurrency. num_gpus=1, num_cpus=..., concurrency=N, and batch_size=... tell Daft how to schedule the UDF across the cluster's GPUs and how big each batch is.
Typed output. The UDF declares its return_dtype (e.g. embedding(float32, 512)), so its output is a proper typed column the rest of the pipeline treats like any other.

The failure modes senior engineers pre-empt.

Loading the model per row/batch. Putting model construction in the call path re-instantiates it constantly. Mitigation: a class UDF that loads the model in the initializer, once per worker.
Downloading before filtering. Fetching every image and then filtering wastes the most expensive step. Mitigation: filter first so .url.download() runs only on surviving rows.
Unbounded object columns. Overusing DataType.python() boxes values and kills vectorization. Mitigation: prefer native image/tensor/embedding types; reserve python() for genuinely unstructured values.

Common interview probes on multimodal columns.

"How does Daft represent an image?" — as a typed image column, decoded in-engine, resizable and encodable with native ops.
"How do you download and decode a column of URLs?" — .url.download() then .image.decode(), chained lazily and parallelized.
"How do you run a GPU model over a column?" — a class-based UDF loading the model once per worker, with num_gpus/concurrency/batch_size.
"What produces an embedding column?" — a UDF with return_dtype=DataType.embedding(...).

Worked example — URL download and image decode as expressions

Detailed explanation. The starting point of most multimodal pipelines is a column of URLs. Turn it into decoded, resized images with a single lazy expression chain, tolerant of bad links.

The input. A string column image_url.
The chain. download → decode → resize.
The resilience. on_error="null" keeps a dead URL from failing the job.

Question. Convert a column of image URLs into a typed, resized image column, dropping rows whose download or decode fails.

Input.

Column	Before	After the chain
`image_url`	string URL	(unchanged)
`bytes`	—	binary (downloaded)
`image`	—	typed `image`
`thumb`	—	`image` 224×224

Code.

import daft
from daft import col

df = daft.from_pydict({
    "id": [1, 2, 3],
    "image_url": [
        "s3://imgs/a.jpg",
        "https://example.com/b.png",
        "s3://imgs/broken.jpg",     # will fail -> null, not a crash
    ],
})

df = (
    df
    .with_column("bytes", col("image_url").url.download(on_error="null"))
    .with_column("image", col("bytes").image.decode(on_error="null"))
    .with_column("thumb", col("image").image.resize(224, 224))
    .where(col("thumb").not_null())     # drop rows that failed download/decode
)
df.show()

Step-by-step explanation.

.url.download(on_error="null") fetches each URL — S3 and HTTPS alike — into a binary bytes column; the engine parallelizes the fetches, so this is not a serial Python loop, and a broken URL yields null instead of raising.
.image.decode(on_error="null") turns the bytes into a typed image column; from this point the engine knows the column is an image and can operate on it natively. A byte blob that is not a valid image also becomes null.
.image.resize(224, 224) runs as a native image op over the decoded column — the kind of operation you would otherwise do in a per-row Python loop with PIL, now vectorized and inside the plan.
.where(col("thumb").not_null()) drops the rows whose download or decode failed, so the bad broken.jpg row is filtered out cleanly and the downstream steps only ever see valid images.
Because the whole chain is lazy, if an earlier where had filtered the DataFrame to 100 rows, only those 100 URLs would be downloaded — the engine never fetches images for rows a filter will discard, which is the multimodal payoff of laziness.

Output.

id	download	decode	result
1	ok	ok	thumb kept
2	ok	ok	thumb kept
3	null (broken)	null	dropped by `not_null`

Rule of thumb. Build media pipelines as lazy expression chains — .url.download() → .image.decode() → .image.resize() — with on_error="null" so bad inputs become nulls you filter out rather than job-killing exceptions. Keep the download late in the plan so filters upstream shrink the set of URLs you actually fetch.

Worked example — a class-based GPU UDF that embeds images

Detailed explanation. The expensive step in a multimodal pipeline is running a model on the GPU, and the right pattern is a class UDF that loads the model once per worker and processes batches. Build an image-embedding UDF.

The model. An image encoder loaded once, on the GPU.
The UDF. A class with an initializer (load) and a call (embed a batch).
The schedule. num_gpus, concurrency, and batch_size control placement.

Question. Implement a GPU image-embedding UDF that loads the model once per worker and returns a typed embedding column.

Input.

Piece	Value
Input column	`thumb` (typed image)
Output dtype	`embedding(float32, 512)`
Model load	once, in the initializer
Resources	`num_gpus=1`, `concurrency=4`, `batch_size=64`

Code.

import daft
from daft import col, DataType

@daft.udf(
    return_dtype=DataType.embedding(DataType.float32(), 512),  # typed output column
    num_gpus=1,          # each replica gets a GPU
    concurrency=4,       # 4 GPU replicas across the cluster
    batch_size=64,       # process 64 images per call
)
class EmbedImages:
    def __init__(self):
        # Runs ONCE per replica — the expensive model load, not per batch.
        import torch
        self.torch = torch
        self.model = load_encoder().cuda().eval()

    def __call__(self, images):
        # `images` is a batch (Series). Convert, run on GPU, return one vector per row.
        batch = preprocess(images.to_pylist())                # -> tensor on GPU
        with self.torch.no_grad():
            vecs = self.model(batch).cpu().numpy()
        return list(vecs)                                     # len == len(images)

df = daft.read_parquet("s3://corpus/*.parquet")
df = df.with_column("thumb", col("url").url.download().image.decode().image.resize(224, 224))
df = df.with_column("embedding", EmbedImages(col("thumb")))   # typed embedding column
df.write_parquet("s3://corpus/embeddings/")

Step-by-step explanation.

The decorator declares the UDF's contract: a return_dtype of embedding(float32, 512) (so the output is a real typed column), plus scheduling hints — num_gpus=1 per replica, concurrency=4 replicas, and a batch_size of 64 images per call.
The initializer runs once per replica and loads the encoder onto the GPU. This is the whole reason to use a class UDF: a plain function UDF would reload the model on every batch, wasting the most expensive setup repeatedly.
The call method receives a batch of images as a Series, not one row — so preprocessing and the forward pass are done on 64 images at a time, amortizing Python-boundary overhead and keeping the GPU fed.
Applying EmbedImages(col("thumb")) inside with_column makes the embedding a typed column of the DataFrame; Daft schedules the four GPU replicas across the Ray cluster automatically because it knows each needs a GPU.
Because the output dtype is a real embedding, downstream steps — writing to Parquet, feeding a vector index, computing similarity — treat it like any typed column, with no boxing or ad-hoc serialization. Model setup happens once per worker; inference happens in batches; the rest of the pipeline stays lazy and distributed.

Output.

Aspect	Function UDF (naive)	Class UDF (correct)
Model load	per batch	once per replica
Batching	per row risk	`batch_size` images/call
GPU placement	manual	`num_gpus`/`concurrency`
Output column	boxed	typed `embedding(512)`

Rule of thumb. Run models over a column with a class-based UDF: load the model once in the initializer, process batches in the call, and declare num_gpus, concurrency, batch_size, and a typed return_dtype. It is the difference between reloading a model thousands of times and loading it once per worker — and it keeps the embedding a first-class column.

Worked example — tensors and embeddings as typed columns

Detailed explanation. Once media is embedded, the vectors are not opaque blobs — they are typed embedding/tensor columns you can operate on and store efficiently. Show working with embedding columns for a similarity computation.

The type. embedding(float32, 512) — a fixed-length vector column.
The operations. Cosine similarity against a query vector, as an expression.
The storage. Written to Parquet as a typed, compact column.

Question. Given an embedding column, compute a similarity score against a query vector and keep the top matches, using typed operations.

Input.

Column	Type	Role
`id`	int64	key
`embedding`	`embedding(float32,512)`	item vector
`score`	float32	cosine similarity
(query)	length-512 vector	search vector

Code.

import daft
from daft import col, DataType, lit
import numpy as np

df = daft.read_parquet("s3://corpus/embeddings/*.parquet")   # has an `embedding` column

query = np.random.rand(512).astype("float32")                # a search vector

# Cosine similarity as a batch UDF over the typed embedding column.
@daft.udf(return_dtype=DataType.float32())
def cosine_to_query(embs, q):
    mat = np.stack(embs.to_pylist())                         # (n, 512)
    qn = q / (np.linalg.norm(q) + 1e-8)
    mn = mat / (np.linalg.norm(mat, axis=1, keepdims=True) + 1e-8)
    return list((mn @ qn).astype("float32"))

df = df.with_column("score", cosine_to_query(col("embedding"), query))
top = df.sort(col("score"), desc=True).limit(10)             # nearest 10
top.select(col("id"), col("score")).show()

Step-by-step explanation.

The embedding column read from Parquet is typed and compact — Daft stored the 512-float vectors as a fixed-size column, not as boxed Python lists, so the read is efficient and the type is preserved end to end.
cosine_to_query is a batch UDF: it receives the whole embedding batch as a Series, stacks it into an (n, 512) matrix once, and computes all similarities with a single vectorized matrix-vector product — not a per-row Python loop.
The UDF's return_dtype=DataType.float32() makes score a proper numeric column, so the subsequent sort and limit are ordinary typed operations the engine optimizes like any tabular query.
sort(col("score"), desc=True).limit(10) uses limit pushdown to keep only the top matches, so even a large corpus produces a small result — the embedding search rides the same lazy-plan machinery as tabular analytics.
The point is that embeddings are data, not an afterthought: because they are a first-class column type, similarity, ranking, and storage are all expressible in the DataFrame, and the same code scales from a laptop test to a distributed search over the whole corpus on Ray.

Output.

id	score	rank
8842	0.94	1
1195	0.91	2
…	…	…
402	0.78	10

Rule of thumb. Treat embeddings and tensors as typed columns, not blobs: store them as embedding/tensor dtypes for compact I/O, compute similarity and ranking as vectorized batch UDFs and expressions, and let limit pushdown keep the result small. The whole search pipeline stays in the DataFrame and scales with the runner.

Senior interview question on building a multimodal Daft pipeline

A senior interviewer might ask: "Build a multimodal pipeline over a table of image URLs: download and decode the images, resize them, run a GPU model to produce embeddings, and write the vectors back — resilient to broken URLs, loading the model once per worker, and keeping the expensive download and inference off rows you will discard. Explain the column types, the UDF pattern, and why this belongs inside the DataFrame rather than in external Python glue."

Solution Using typed media columns, a resilient download-decode chain, and a class GPU UDF

# 1 — Typed media chain: filter FIRST, then download/decode only surviving rows.
import daft
from daft import col, DataType

df = daft.read_parquet("s3://catalog/products/*.parquet")
df = df.where(col("in_stock") == True)                       # shrink the set BEFORE download
df = (df
      .with_column("bytes", col("image_url").url.download(on_error="null"))
      .with_column("image", col("bytes").image.decode(on_error="null"))
      .with_column("thumb", col("image").image.resize(224, 224))
      .where(col("thumb").not_null()))                       # drop broken URLs/decodes

# 2 — Class GPU UDF: model loaded once per worker, batched inference, typed output.
@daft.udf(return_dtype=DataType.embedding(DataType.float32(), 512),
          num_gpus=1, concurrency=8, batch_size=64)
class Embed:
    def __init__(self):
        import torch
        self.torch = torch
        self.model = load_encoder().cuda().eval()            # ONCE per replica
    def __call__(self, images):
        batch = preprocess(images.to_pylist())
        with self.torch.no_grad():
            return list(self.model(batch).cpu().numpy())

df = df.with_column("embedding", Embed(col("thumb")))

# 3 — Drop heavy intermediates and write typed vectors back to the lake.
df = df.exclude("bytes", "image")                            # don't persist raw bytes/images
df.select(col("sku"), col("embedding")).write_parquet("s3://catalog/embeddings/")
# Same program runs local (native) or distributed (Ray) — just set the runner.

Step-by-step trace.

Layer	Component	Purpose
Filter-first	`where(in_stock)` before download	fetch only rows you keep
Media types	`url.download` → `image.decode` → `resize`	typed image column in-engine
Resilience	`on_error="null"` + `not_null()`	broken URLs drop, no crash
Model	class UDF, load in `__init__`	model once per worker
Inference	`num_gpus`/`concurrency`/`batch_size`	batched GPU scheduling
Output	typed `embedding` to Parquet	compact, first-class vectors

After deployment, the pipeline filters to in-stock products before downloading, so no bandwidth or GPU time is spent on rows that will be discarded; the surviving URLs are downloaded, decoded, and resized as typed image columns, with broken links turning to nulls that a not_null() filter removes; a class UDF loads the encoder once per GPU replica and embeds images in batches of 64; and the typed embeddings are written back to the lake with the raw bytes and full images excluded. Every step lives in the DataFrame, so the optimizer, streaming, and the local-or-Ray runner all apply — no external media-glue script exists.

Output:

Metric	External glue	Daft multimodal
Image representation	paths + side code	typed `image` column
Download of discarded rows	common	none (filter-first)
Broken URL	crashes / special-cases	null, filtered out
Model loads	per batch	once per worker
Embedding storage	boxed / ad-hoc	typed `embedding` column

Why this works — concept by concept:

Typed media columns — image, bytes, and embedding are real column types, so download, decode, resize, and embed happen inside the plan where the optimizer, streaming, and distribution still apply — not in an external loop the engine cannot see.
Filter-first laziness — because the chain is lazy, filtering to in-stock rows before .url.download() means the expensive fetch and GPU inference run only on rows that survive, the single biggest cost saving in a media pipeline.
Resilient accessors — on_error="null" turns broken URLs and undecodable bytes into nulls a not_null() filter removes, so one bad row never fails a multi-million-row job.
Class UDF for models — loading the model in the initializer runs it once per worker while batched calls keep the GPU fed, and a typed return_dtype makes the embedding a first-class column the rest of the pipeline consumes directly.
Cost — one model load per worker, downloads and inference only on kept rows, and compact typed vector storage, versus reloading the model per batch, fetching images you discard, and boxing vectors. The eliminated cost is the wasted bandwidth, GPU time, and glue code of doing media work outside the engine — O(kept rows) instead of O(all rows, reloaded model).

Data transformation
Topic — data-transformation
Data transformation problems on media pipelines and UDFs

Practice →

Data processing Topic — data-processing Data processing problems on embeddings and typed columns

Practice →

5. ML data pipelines — Parquet, Iceberg, batch inference

One engine reads the lakehouse, runs GPU inference at scale, and streams batches into training

The mental model in one line: Daft closes the loop for ML data — it reads the lakehouse (read_parquet, read_iceberg, read_deltalake) from object storage with an IOConfig and full pushdown, runs distributed batch inference as a model UDF over a lazy plan, writes results back as typed columns, and then streams the same DataFrame straight into training via to_torch_iter_dataset() / to_ray_dataset() / iter_rows() — so the read-transform-infer-train path that usually spans three tools and a fragile handoff becomes one program that scales from a laptop to a Ray cluster. The data engine and the ML data loader are the same DataFrame.

Reading the lakehouse.

Formats. daft.read_parquet(...), daft.read_csv(...), daft.read_json(...) for files, and daft.read_iceberg(table) / daft.read_deltalake(...) for table formats that bring schema, partitioning, and snapshot semantics.
Object storage + IOConfig. An IOConfig (with an S3Config, GCS, or Azure config) carries credentials and region, so read_parquet("s3://...", io_config=io) reads directly from the lake with the right auth.
Pushdown into table formats. Reading Iceberg/Delta lets Daft use partition and file pruning plus column projection, so a filtered training-split read scans only the relevant files — the same pushdown discipline as section 2.
Globs and many files. A glob (s3://lake/x/*.parquet) sets the initial partitioning; a well-laid-out dataset of balanced files parallelizes cleanly across the cluster.

Batch inference at scale.

A model UDF over a distributed DataFrame. The same class-UDF pattern from section 4 runs a model across the cluster's GPUs, so inference over a billion rows is a with_column on a lazy, streamed plan.
Write results back. df.write_parquet(...) (or write_iceberg/write_deltalake) persists predictions/embeddings as typed columns next to the inputs, closing the loop back into the lake.
Streaming keeps memory bounded. Because the plan streams, inference over a dataset far larger than memory produces and writes results incrementally rather than buffering everything.
Resource control. num_gpus, concurrency, and batch_size size the inference stage to the cluster, and backpressure keeps the GPU fed without overrunning memory.

Feeding training.

PyTorch. df.to_torch_iter_dataset() yields an IterableDataset you wrap in a DataLoader, streaming batches into a training loop without materializing the dataset — the out-of-core property carried into training.
Ray Train / Ray Data. df.to_ray_dataset() hands the DataFrame to Ray's training ecosystem for distributed data-parallel training.
Row iteration. df.iter_rows() / df.iter_partitions() stream results for custom consumers; to_pandas()/to_arrow() materialize when a step genuinely needs the whole thing in memory.
Preprocess in the DataFrame. Decode, resize, normalize, and tokenize as Daft expressions/UDFs before handing to the loader, so the GPU training loop receives ready tensors and is never starved by CPU preprocessing.

The failure modes senior engineers pre-empt.

Materializing before training. Calling .to_pandas() on a huge dataset to feed a loader defeats out-of-core and OOMs. Mitigation: stream with to_torch_iter_dataset() / iter_partitions().
CPU-starved GPUs. Doing decode/resize in the training loop leaves the GPU idle waiting on CPU work. Mitigation: push preprocessing into the Daft plan so batches arrive ready.
Reading the whole table for a split. Ignoring partition/column pushdown and scanning the full lake for one training split. Mitigation: filter on partition columns and select only needed columns so the read prunes.

Common interview probes on ML pipelines.

"How does Daft read from S3/Iceberg?" — read_parquet/read_iceberg with an IOConfig, using partition/column pushdown.
"How do you run batch inference at scale?" — a class model UDF over the distributed DataFrame, writing typed results back to the lake.
"How do you feed training without OOM?" — stream via to_torch_iter_dataset()/to_ray_dataset(), preprocessing in the DataFrame first.
"Where does preprocessing belong?" — in the Daft plan, so the GPU loop gets ready tensors and stays fed.

Worked example — read Parquet and Iceberg from S3 with pushdown

Detailed explanation. The pipeline starts at the lake. Read a training split from Parquet and from an Iceberg table with credentials and pushdown, so only the needed files and columns are scanned.

The auth. An IOConfig with an S3Config.
The formats. read_parquet for files, read_iceberg for a table.
The pruning. Filter on the partition column, select only needed columns.

Question. Read a training split from S3 Parquet and from an Iceberg table, using an IOConfig and pushdown so the scan is minimal.

Input.

Aspect	Value
Auth	`IOConfig(s3=S3Config(region_name=...))`
Parquet	`read_parquet("s3://lake/x/*.parquet", io_config=io)`
Iceberg	`read_iceberg(table)` from a catalog
Pushdown	`where(split=='train')` + `select(...)`

Code.

import daft

# 1 — Credentials/region for object storage, passed to the reader.
io = daft.io.IOConfig(s3=daft.io.S3Config(region_name="us-east-1"))

# 2 — Parquet on S3, with partition + projection pushdown.
df = daft.read_parquet("s3://lake/images/*.parquet", io_config=io)
train = (df
         .where(daft.col("split") == "train")        # partition prune
         .select(daft.col("image_url"), daft.col("label")))   # 2 columns only

# 3 — Iceberg table via a catalog (schema, partitioning, snapshots come for free).
from pyiceberg.catalog import load_catalog
catalog = load_catalog("glue")
table = catalog.load_table("ml.image_labels")
ice = (daft.read_iceberg(table)
       .where(daft.col("split") == "train")          # Iceberg partition pruning
       .select(daft.col("image_url"), daft.col("label")))

train.show(3)

Step-by-step explanation.

The IOConfig carries the S3 region and (implicitly) credentials, so the reader authenticates against object storage; the same object plugs into every read_* call, keeping auth in one place rather than scattered per read.
read_parquet(..., io_config=io) is lazy, and the where(split=='train') becomes a partition filter while the select becomes a projection — so a lake partitioned by split opens only the train partition and reads only two columns.
read_iceberg(table) reads through the table format, which means Daft uses Iceberg's metadata for partition and file pruning and gets the table's schema and current snapshot — a more governed read than raw file globbing.
Both reads apply the same pushdown discipline: filter on the partition column and select the minimum columns, so whether the source is raw Parquet or an Iceberg table, the scan touches only the training split and the two needed fields.
The senior habit is to let the table format do the pruning where it exists: Iceberg/Delta bring partitioning and statistics that make pushdown precise, so reading a governed table is both cleaner and cheaper than scanning a directory of files by hand.

Output.

Source	Files opened	Columns read
Parquet glob	only `split=train`	`image_url`, `label`
Iceberg table	pruned by metadata	`image_url`, `label`
whole table	(avoided)	(avoided)

Rule of thumb. Read the lake with an IOConfig for auth and let pushdown prune the scan: filter on partition columns and select only what you need, and prefer table formats (Iceberg/Delta) where they exist so partition and file pruning are precise. The read is where most ML-pipeline waste hides — minimize it first.

Worked example — distributed batch inference writing back to the lake

Detailed explanation. With the read minimized, run a model over the whole dataset on the cluster and persist the predictions. This is the section-4 UDF pattern applied at ML-pipeline scale.

The scale. A billion rows, GPU inference, distributed on Ray.
The UDF. A class model UDF, model loaded once per worker.
The sink. write_parquet back to the lake, typed predictions.

Question. Run a GPU classification model over a distributed Daft DataFrame and write the labelled results back to the lakehouse.

Input.

Piece	Value
Runner	Ray (distributed)
Model UDF	class, `num_gpus=1`, `concurrency=8`
Output	`label` (string) column
Sink	`write_parquet("s3://.../labeled/")`

Code.

import daft
from daft import col, DataType

daft.context.set_runner_ray(address="ray://head:10001")      # distribute inference

io = daft.io.IOConfig(s3=daft.io.S3Config(region_name="us-east-1"))
df = daft.read_parquet("s3://lake/images/*.parquet", io_config=io)
df = df.with_column("img", col("url").url.download().image.decode().image.resize(224, 224))

@daft.udf(return_dtype=DataType.string(), num_gpus=1, concurrency=8, batch_size=32)
class Classify:
    def __init__(self):
        import torch
        self.torch = torch
        self.model, self.labels = load_classifier()          # once per replica
        self.model = self.model.cuda().eval()
    def __call__(self, images):
        x = preprocess(images.to_pylist())
        with self.torch.no_grad():
            idx = self.model(x).argmax(dim=1).cpu().tolist()
        return [self.labels[i] for i in idx]

df = df.with_column("label", Classify(col("img")))
df.exclude("img").write_parquet("s3://lake/images_labeled/", io_config=io)   # stream back

Step-by-step explanation.

Selecting the Ray runner makes the whole plan distributed, so the download-decode-resize and the inference stage all run across the cluster's workers and GPUs — the same code that would run on one machine with the native runner.
The Classify UDF loads the model and its labels once per replica in the initializer, then runs batched inference in the call; concurrency=8 and num_gpus=1 tell Daft to place eight GPU replicas across the cluster.
with_column("label", Classify(col("img"))) adds the prediction as a typed string column on the lazy plan — inference is just another column, not a separate job with its own orchestration.
write_parquet(..., io_config=io) streams the labelled rows back to the lake incrementally; because the plan streams, a billion-row inference never buffers the whole dataset — it produces and writes results as it goes, bounded in memory.
exclude("img") drops the heavy decoded-image column before the write, so the persisted output carries the compact inputs and the new label rather than re-storing every image — a deliberate choice to keep the sink cheap.

Output.

Stage	Behavior
read	pushed-down scan from S3
decode/resize	typed image column, distributed
inference	8 GPU replicas, batched
write	streamed back, `img` excluded
memory	bounded (streaming)

Rule of thumb. Run batch inference as a model UDF over the distributed DataFrame and stream the typed results straight back to the lake with write_parquet/write_iceberg, excluding heavy intermediate columns before the write. Inference becomes one with_column on a lazy, streamed plan — no separate serving job, no buffering the dataset.

Worked example — streaming a Daft DataFrame into a PyTorch training loop

Detailed explanation. The last hop is training. Stream the preprocessed DataFrame into a PyTorch loader without materializing it, doing all the CPU preprocessing in Daft so the GPU stays fed.

The preprocessing. decode, resize, normalize — as Daft expressions.
The bridge. to_torch_iter_dataset() → DataLoader.
The property. Streamed, out-of-core, GPU never starved.

Question. Feed a preprocessed Daft DataFrame into a PyTorch training loop as a streaming iterable dataset, with preprocessing done in the DataFrame.

Input.

Piece	Value
Preprocess	in Daft: decode → resize → normalize
Bridge	`to_torch_iter_dataset()`
Loader	`torch.utils.data.DataLoader(..., batch_size=256)`
Property	streaming, no full materialization

Code.

import daft
from daft import col
import torch

# 1 — All preprocessing in the DataFrame, so the training loop gets ready tensors.
df = daft.read_parquet("s3://lake/train/*.parquet")
df = (df
      .with_column("img", col("url").url.download().image.decode().image.resize(224, 224))
      .with_column("x", col("img").image.to_tensor())      # typed tensor column
      .select(col("x"), col("label")))

# 2 — Stream into PyTorch WITHOUT materializing the dataset in memory.
torch_ds = df.to_torch_iter_dataset()                       # IterableDataset (streamed)
loader = torch.utils.data.DataLoader(torch_ds, batch_size=256, num_workers=4)

# 3 — Ordinary training loop; batches arrive pre-decoded and pre-resized.
model = build_model().cuda().train()
opt = torch.optim.Adam(model.parameters())
for batch in loader:
    x = batch["x"].cuda(non_blocking=True)
    y = batch["label"].cuda(non_blocking=True)
    loss = model.loss(model(x), y)
    opt.zero_grad(); loss.backward(); opt.step()

Step-by-step explanation.

All the heavy per-sample work — download, decode, resize, convert to a tensor — is expressed in the Daft plan, so it runs in the streaming Rust engine (and can be distributed), not inside the Python training loop where it would bottleneck the GPU.
to_torch_iter_dataset() returns an IterableDataset that streams rows from the plan; the dataset is never fully materialized, so training over a corpus far larger than memory works exactly like the out-of-core scans from section 3.
The DataLoader wraps that iterable with batching and worker prefetching, so batches of 256 ready tensors are staged ahead of the GPU — the loader pulls from Daft's stream rather than from an in-memory array.
The training loop is completely ordinary: because x arrives already decoded and resized, each step moves a ready batch to the GPU and does a forward/backward pass, and the GPU is never left idle waiting on CPU preprocessing.
The senior payoff: read, preprocess, infer, and train are one program in one framework — no export-to-a-second-tool handoff, no separate preprocessing job whose output drifts from the training code, and out-of-core streaming holds all the way into the loop.

Output.

Concern	Preprocess in loop	Preprocess in Daft
GPU utilization	starved (CPU-bound)	fed (ready tensors)
Dataset in memory	often materialized	streamed
Larger-than-memory	OOM risk	works (out-of-core)
Code paths	two tools	one DataFrame

Rule of thumb. Do all preprocessing in the Daft plan and stream into training with to_torch_iter_dataset() (or to_ray_dataset()), never materializing a big dataset to hand it over. The GPU receives ready tensors and stays fed, out-of-core streaming reaches into the training loop, and read-preprocess-train is one program instead of a fragile multi-tool handoff.

Senior interview question on an end-to-end Daft ML pipeline

A senior interviewer might ask: "Design the full data path for training and batch-scoring an image model: read a training split from an Iceberg/Parquet lake on S3, preprocess and embed on GPUs, write embeddings back, and stream batches into training — all bounded in memory, minimizing what you read, loading models once per worker, and running the same code locally and on a Ray cluster. Walk the pieces and justify each choice."

Solution Using pushed-down lake reads, a GPU model UDF, and streamed training

# 1 — Minimal read: IOConfig auth + partition/column pushdown from the lake.
import daft
from daft import col, DataType

io = daft.io.IOConfig(s3=daft.io.S3Config(region_name="us-east-1"))
df = (daft.read_parquet("s3://lake/images/*.parquet", io_config=io)
      .where(col("split") == "train")               # partition prune
      .select(col("url"), col("label")))            # projection prune

# 2 — Preprocess + GPU embedding in the plan; model loaded once per worker.
df = df.with_column("img", col("url").url.download(on_error="null")
                                      .image.decode(on_error="null")
                                      .image.resize(224, 224)).where(col("img").not_null())

@daft.udf(return_dtype=DataType.embedding(DataType.float32(), 512),
          num_gpus=1, concurrency=8, batch_size=64)
class Embed:
    def __init__(self):
        import torch; self.torch = torch
        self.model = load_encoder().cuda().eval()    # once per replica
    def __call__(self, images):
        with self.torch.no_grad():
            return list(self.model(preprocess(images.to_pylist())).cpu().numpy())

df = df.with_column("embedding", Embed(col("img")))
df.select(col("label"), col("embedding")).write_parquet("s3://lake/embeddings/", io_config=io)

# 3 — Stream the SAME DataFrame into training; run local or on Ray unchanged.
import torch
daft.context.set_runner_ray(address="ray://head:10001")   # or native locally
torch_ds = df.select(col("embedding"), col("label")).to_torch_iter_dataset()
loader = torch.utils.data.DataLoader(torch_ds, batch_size=256, num_workers=4)
for batch in loader:                                 # ready tensors, streamed, out-of-core
    train_step(batch)

Step-by-step trace.

Layer	Component	Purpose
Read	IOConfig + pushdown	scan only train split, 2 columns
Preprocess	typed media chain in-plan	decode/resize inside the engine
Embed	class GPU UDF	model once/worker, batched
Persist	`write_parquet` back	typed embeddings to the lake
Train	`to_torch_iter_dataset()`	streamed batches, GPU fed
Scale	runner switch	same code local or on Ray

After deployment, the pipeline reads only the training split and two columns from the lake (partition and projection pushdown), downloads and decodes images as typed columns with broken links dropped, and embeds them with a class UDF that loads the encoder once per GPU replica and batches 64 images per call. The typed embeddings are streamed back to the lake, and the same DataFrame is streamed into a PyTorch loader so training receives ready tensors without materializing the corpus. Memory stays bounded end to end, and flipping the runner from native to Ray scales the whole path from a laptop to the cluster with no logic change.

Output:

Metric	Multi-tool handoff	One Daft pipeline
Bytes read for a split	whole table	pruned scan
Media work	external glue	typed columns in-plan
Model loads	per batch/job	once per worker
Dataset for training	materialized/exported	streamed, out-of-core
Local ↔ cluster	separate code	one runner switch

Why this works — concept by concept:

Pushed-down lake reads — an IOConfig for auth plus partition and projection pushdown means the scan touches only the training split and the columns the pipeline uses, so the most expensive step (I/O from object storage) is minimized before anything else runs.
In-plan preprocessing — decode, resize, and embed are typed-column operations inside the lazy plan, so they stream, distribute, and stay off the training loop's critical path instead of starving the GPU from a Python preprocessing step.
Model UDF once per worker — the class UDF loads the encoder in the initializer and batches inference, so a billion-row embed loads the model a handful of times, not per batch, and Daft places the GPU replicas across the cluster.
Streamed training bridge — to_torch_iter_dataset() streams the same DataFrame into the loader without materializing it, carrying the out-of-core property into training so the corpus can dwarf memory.
Cost — a pruned read, in-plan streamed preprocessing, model loads per worker, and a streamed training feed, versus a full scan, an external preprocessing job, per-batch model loads, and a materialized export. The eliminated cost is the I/O, memory, and glue of a three-tool handoff — O(needed data, model-per-worker) instead of O(whole table, reloaded model, materialized dataset).

Data processing
Topic — data-processing
Data processing problems on lakehouse reads and batch inference

Practice →

API integration
Topic — api-integration
API integration problems on connectors and external data sources

Practice →

Cheat sheet — Daft for multimodal & ML data

What Daft is. A Python-first DataFrame on a Rust engine — lazy, columnar (Arrow), multimodal-native, and runnable locally (native runner) or distributed (Ray) from the same code. It fills the seam where Pandas is single-node/eager, Spark is JVM-bound and tabular-only, and Polars is single-machine.
When to reach for it. Big and multimodal: larger-than-memory data with images/tensors/embeddings that must scale. Pandas/Polars for single-node tabular; Spark for JVM-shop tabular ETL; Ray Data for a pure ML block stream; Daft for the intersection (distributed + multimodal + a query optimizer).
Lazy plan + action. Transformations (read_*, where, select, with_column, join, groupby/agg) build a plan and read nothing; actions (.collect(), .show(), .write_parquet(), .to_torch_iter_dataset()) trigger execution. Debug with df.explain(show_all=True) — confirm predicate/projection/limit pushdown landed on the scan.
Optimizer. Predicate pushdown (skip partitions/row groups), projection pushdown (read only needed columns), limit pushdown (stop early), column pruning. Keep filters as native expressions (not opaque UDFs) so they stay pushable; filter/project first, apply expensive UDFs last.
Columnar / Rust. Data is Arrow columnar with a real type system; expressions compile to vectorized Rust kernels, so the heavy work never enters the Python interpreter and there is no JVM.
Runners. daft.context.set_runner_native() for one machine (multithreaded, streaming, out-of-core, spill-to-disk); daft.context.set_runner_ray(address=...) (or DAFT_RUNNER=ray) for a cluster. The DataFrame logic is runner-agnostic — scaling out is a config change, not a rewrite.
Out-of-core. Morsel-driven streaming plus spill-to-disk means peak memory tracks in-flight batch size, not dataset size — a 500 GB job runs on a small box; don't down-sample.
Partitioning. Partitions → tasks. repartition(n, key) to hash-partition both join sides on the join key (key-local join); align the group-by key to reuse the partitioning; size partitions to low hundreds of MB and ~2–4× cluster cores. Fix a straggler (skew) by salting the hot key: split → aggregate → combine.
Multimodal columns. DataType.image() / .tensor() / .embedding(dtype, size) / .python(). col("url").url.download(on_error="null") → .image.decode(on_error="null") → .image.resize(h, w), chained lazily. Filter before download so you never fetch rows you discard.
UDFs. Batch function UDF (@daft.udf(return_dtype=...)) receives a Series; class UDF loads a model once in the initializer and batches in the call — use num_gpus, concurrency, batch_size, and a typed return_dtype. Never load the model per batch.
Lakehouse I/O. read_parquet / read_iceberg / read_deltalake with an IOConfig(s3=S3Config(...)); prune with partition filters + projections. Write results back with write_parquet / write_iceberg, excluding heavy intermediate columns first.
Feed training. Preprocess in the DataFrame, then stream with to_torch_iter_dataset() / to_ray_dataset() / iter_partitions() — never .to_pandas() a huge dataset to hand it over. The GPU gets ready tensors; out-of-core streaming reaches into the training loop.

Frequently asked questions

What is Daft and how is it different from Pandas and Spark?

Daft is an open-source DataFrame with a Python-first API sitting on a Rust execution engine; it is lazy, columnar (Apache Arrow), treats multimodal data as first-class typed columns, and runs either multithreaded on one machine or distributed across a Ray cluster from the same code. The difference from Pandas is scale and laziness: Pandas is eager and single-node, so it materializes intermediates and fails when data outgrows memory, whereas Daft builds a query plan an optimizer rewrites and streams data out-of-core so bigger-than-memory jobs run without down-sampling. The difference from Spark is the engine and the data model: Spark runs on the JVM with a Python serialization boundary and represents data as tabular rows, whereas Daft is Rust-native with no JVM and can hold an image, a tensor, or an embedding as a genuine column — so machine-learning pipelines that Spark forces into external glue stay inside the DataFrame. In short, Daft aims to be as expressive as Pandas, as scalable as Spark, and multimodal in a way neither is.

Why is Daft written in Rust, and what does that buy me?

Writing the engine in Rust means the heavy work — I/O, decoding, filtering, arithmetic, aggregation, joins — runs as native, vectorized code over Arrow columnar buffers, with no per-row Python interpreter overhead and no JVM. Practically, that buys three things: raw speed (SIMD- and cache-friendly kernels on contiguous typed arrays), predictable memory (a streaming, out-of-core engine that spills to disk rather than OOMing), and a clean Python experience (you write Python, but execution never pays the Python-per-row tax that makes Pandas apply and Spark Python UDFs slow). It also avoids the operational weight of a JVM stack — no separate cluster runtime to tune for garbage collection and heap. You still write ordinary Python; Rust is the engine under it, and it is why Daft can be both Python-friendly and fast on large, multimodal data.

What does "multimodal DataFrame" actually mean?

It means the DataFrame's type system includes media and ML types — DataType.image(), DataType.tensor(), DataType.embedding(dtype, size), and a DataType.python() escape hatch — alongside the usual numbers, strings, and timestamps, so an image or a vector is a real column the engine understands rather than a file path you carry around and process elsewhere. Concretely, a URL column becomes bytes with .url.download(), bytes become a typed image with .image.decode(), images are resized with .image.resize(), and a model turns images into an embedding column through a UDF — all as lazy expressions inside the plan, so the optimizer, streaming execution, and distribution apply to the media work just as they do to tabular work. The payoff is that download → decode → embed → store is a few columns in one framework, instead of a tabular DataFrame plus a separate pile of Python glue that the engine cannot see or optimize.

When should I use Daft's local runner vs the Ray runner?

Use the native (local) runner for development, testing, and any job that fits comfortably on one machine — it is a multithreaded, streaming, out-of-core Rust engine, so it already handles data larger than memory on a single node by spilling to disk, and it has the lowest overhead and simplest operations. Switch to the Ray runner when the work exceeds one machine: when you need more aggregate CPU/GPU than a single node has, when the dataset or the inference throughput demands many workers, or when you are already running on a Ray cluster. The important part is that the DataFrame logic does not change — you select the runner with daft.context.set_runner_native() or daft.context.set_runner_ray(address=...) (or the DAFT_RUNNER environment variable), so the same program you tested locally is the one that runs distributed. A good workflow is to build and debug on the native runner against sampled data, then flip to Ray for the full-scale run.

Daft vs Ray Data vs Polars — which do I pick?

Pick by what the workload needs on three axes: distribution, multimodal support, and whether you want a full query optimizer. Polars is a superb single-machine Rust DataFrame with a streaming engine — choose it for fast tabular work that fits on one box, but it does not distribute across a cluster and is not built around multimodal columns. Ray Data is a distributed dataset for ML preprocessing on Ray — choose it when you want a streaming block dataset feeding distributed training and are comfortable writing the transformation logic yourself, but it is a lower-level dataset rather than a DataFrame with a SQL-grade optimizer. Daft is the pick when you need the intersection: a distributed, Python-first DataFrame with a query optimizer and first-class multimodal columns, running local or on Ray from one code path — which is exactly the combination image/embedding pipelines at scale require. They also compose: Daft runs on Ray and can hand off to Ray Data / Ray Train, so "Daft or Ray Data" is often "Daft feeding Ray."

How does Daft feed a machine-learning training or batch-inference job?

For batch inference, you express the model as a UDF — typically a class UDF that loads the model once per worker in its initializer and processes batches in its call, tagged with num_gpus, concurrency, and batch_size — and apply it with with_column over a distributed DataFrame, then write the typed predictions or embeddings back to the lake with write_parquet/write_iceberg; because the plan streams, this scales past memory and runs across the cluster's GPUs. For training, you do all preprocessing (decode, resize, normalize, tokenize) as Daft expressions so the GPU receives ready tensors, then stream the DataFrame into the trainer with to_torch_iter_dataset() for a PyTorch DataLoader or to_ray_dataset() for Ray Train — without materializing the dataset, so the out-of-core property carries into the training loop. The result is that read, preprocess, infer, and train are one program in one framework rather than a fragile handoff between a data tool and an ML tool, and the same code runs locally or on Ray.

Practice on PipeCode

Drill the data processing practice library → for the DataFrame, large-dataset, and columnar-transform problems that Daft's lazy engine makes concrete.
Rehearse pipeline shaping on the data transformation practice library → for the download-decode-embed and UDF patterns that turn raw media into typed columns.
Sharpen the architecture axis with the system design practice library → for the runner, partitioning, shuffle, and precompute trade-offs a distributed DataFrame must get right.
Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the lazy-evaluation, pushdown, out-of-core, and batch-inference patterns against real graded inputs — DataFrames, transformations, optimization, and streaming.

Lock in Daft muscle memory

Docs explain the Daft API. PipeCode drills explain the decision — when `lazy evaluation` plus pushdown beats an eager scan, when a `multimodal` column belongs inside the DataFrame instead of external glue, when to switch from the native runner to `Ray`, and when to stream into training instead of materializing. Pipecode.ai is Leetcode for Data Engineering — DataFrame and pipeline practice tuned for the production trade-offs data and ML engineers actually face.

Practice data processing problems →
Practice data transformation problems →

1. Why Daft — the gap it fills

The consumption gap — Pandas is single-node, Spark can't hold an image, and ad-hoc Ray re-invents a query engine

Worked example — the DataFrame-engine decision table

Worked example — what interviewers actually probe

Worked example — Daft vs Spark vs Ray Data vs Polars

Senior interview question on choosing a DataFrame engine for multimodal ML

Solution Using a lazy, columnar, multimodal Daft DataFrame that scales from native to Ray

2. The execution model — lazy DataFrame, optimizer, columnar

Transformations build a plan; the optimizer rewrites it; the Rust engine runs it only when you collect

Worked example — build a lazy plan and read .explain()

Worked example — predicate and projection pushdown into a Parquet scan

Worked example — typed columns and expressions over Arrow

Senior interview question on Daft's lazy execution and optimizer

Solution Using a lazy plan, pushdown optimization, columnar execution, and .explain()

3. Distributed execution on Ray — local to cluster

The same lazy plan runs multithreaded on one machine or distributed across a Ray cluster

Worked example — same code, native runner to Ray runner

Worked example — streaming a larger-than-memory scan out-of-core

Worked example — partitioning for a distributed group-by and join

Senior interview question on scaling a Daft pipeline onto Ray

Solution Using one runner switch, streaming execution, and key-aligned partitioning

4. Multimodal columns — URLs, images, embeddings

Images, tensors, and embeddings are first-class typed columns — transformed by expressions and UDFs

Worked example — URL download and image decode as expressions

Worked example — a class-based GPU UDF that embeds images

Worked example — tensors and embeddings as typed columns

Senior interview question on building a multimodal Daft pipeline

Solution Using typed media columns, a resilient download-decode chain, and a class GPU UDF

5. ML data pipelines — Parquet, Iceberg, batch inference

One engine reads the lakehouse, runs GPU inference at scale, and streams batches into training

Worked example — read Parquet and Iceberg from S3 with pushdown

Worked example — distributed batch inference writing back to the lake

Worked example — streaming a Daft DataFrame into a PyTorch training loop

Senior interview question on an end-to-end Daft ML pipeline

Solution Using pushed-down lake reads, a GPU model UDF, and streamed training

Cheat sheet — Daft for multimodal & ML data

Frequently asked questions

What is Daft and how is it different from Pandas and Spark?

Why is Daft written in Rust, and what does that buy me?

What does "multimodal DataFrame" actually mean?

When should I use Daft's local runner vs the Ray runner?

Daft vs Ray Data vs Polars — which do I pick?

How does Daft feed a machine-learning training or batch-inference job?

Practice on PipeCode

Lock in Daft muscle memory

Worked example — build a lazy plan and read `.explain()`

Solution Using a lazy plan, pushdown optimization, columnar execution, and `.explain()`