Intermediate
You have a pipeline that works perfectly on 10,000 rows. Then your dataset grows to 50 million rows and Pandas starts swapping to disk, Polars runs out of memory, and your laptop fan sounds like a helicopter. You need a DataFrame tool that can seamlessly scale from your laptop to a distributed Ray cluster — without rewriting your entire codebase. That is exactly what Daft is built for.
Daft is an open-source DataFrame library written in Rust with a Python API, developed by Eventual. It uses lazy evaluation, columnar execution, and a query optimizer modelled on database engines. On a single machine it already outperforms Pandas significantly. When you attach a Ray cluster, the same code runs distributed across hundreds of cores with no changes. It also handles multimodal data natively — URLs, images, embeddings, and documents sit alongside numeric columns in the same DataFrame.
In this article you will learn how to install Daft, create DataFrames from Python dicts, CSV, and Parquet files, filter and aggregate data, use the URL and image types for multimodal workflows, and scale a job to a Ray cluster. By the end you will have a complete working example that reads a large synthetic dataset, runs aggregations, and shows how to attach Ray with a single line of code.
How To Use Daft: Quick Example
If you are in a hurry, here is a minimal working example that shows the core Daft workflow — create a DataFrame, filter it, and collect the result:
# quick_daft.py
import daft
# Create a DataFrame from a Python dictionary
df = daft.from_pydict({
"product": ["widget", "gadget", "gizmo", "widget", "gadget"],
"region": ["east", "west", "east", "west", "east"],
"sales": [120, 85, 200, 95, 310],
})
# Filter: only rows where sales > 100
high_sales = df.where(df["sales"] > 100)
# Select specific columns and collect the result
result = high_sales.select("product", "region", "sales").collect()
print(result)
Output:
+---------+--------+-------+
| product | region | sales |
| Utf8 | Utf8 | Int64 |
+=========+========+=======+
| widget | east | 120 |
+---------+--------+-------+
| gizmo | east | 200 |
+---------+--------+-------+
| gadget | east | 310 |
+---------+--------+-------+
(Showing first 3 of 3 rows)
Two things to notice. First, daft.from_pydict() builds a lazy DataFrame — no data is processed yet. Second, .collect() is what actually triggers execution and materializes the result. This lazy-first design means Daft can plan and optimize the whole query before touching a single byte of data.
The sections below go deeper: reading from files, grouping, aggregating, handling URLs as first-class data types, and scaling to a Ray cluster.
What Is Daft and Why Use It?
Daft is a distributed query engine that presents itself as a DataFrame API. Under the hood it compiles your Python method calls into a logical query plan, optimizes it (pushing down filters, eliminating unnecessary column reads), and then executes it either locally or on a distributed Ray cluster.
The key design choices that make Daft different from Pandas and Polars:
- Lazy by default. Calling
df.where(...)does not run anything. You chain transformations, and execution only happens when you call.collect(),.show(), or write to a sink. This lets the optimizer reorder and merge operations. - Distributed-first. Attach Ray with
daft.context.set_runner_ray()and the same code distributes across any cluster. No rewrite needed. - Multimodal columns. Daft has built-in URL, Image, and Embedding types. You can store image URLs in a column, call
.url.download(), and then.image.decode()— all in a distributed pipeline. - Rust core. The query engine is written in Rust, giving it performance close to compiled code even for single-machine workloads.
Here is how Daft compares to other Python DataFrame libraries:
| Feature | Pandas | Polars | Daft |
|---|---|---|---|
| Execution model | Eager | Lazy + eager | Lazy |
| Distributed | No | No | Yes (Ray) |
| Multimodal types | No | No | Yes (URL, Image, Embedding) |
| Core language | Python/C | Rust | Rust |
| Query optimizer | No | Yes | Yes |
| Best for | Small data, legacy code | Fast single-node analytics | Big data + multimodal pipelines |
If your data fits in RAM and you do not need image or URL handling, Polars is probably a better fit. If you need to scale beyond one machine or handle multimodal payloads alongside tabular data, Daft is the right tool.
Installing Daft
Install the base package with pip. The [all] extra adds integrations for Ray, AWS S3, Delta Lake, and Apache Iceberg:
# install_daft.sh
pip install daft # base install, local runner only
pip install "daft[all]" # recommended: includes Ray, S3, Delta Lake, Iceberg
Output:
Successfully installed daft-0.4.x ...
Verify the install by importing and checking the version:
# verify_daft.py
import daft
print(daft.__version__)
Output:
0.4.x
If you see the version string, you are ready. The base install is enough to follow every example in this article. You only need the [all] extra when you connect to Ray or read from cloud storage.
Creating DataFrames in Daft
Daft can ingest data from Python objects, CSV files, Parquet files, JSON files, and cloud storage. Here are the most common entry points.
From a Python Dictionary
The fastest way to get started is daft.from_pydict(). It converts a plain Python dictionary of lists into a Daft DataFrame, inferring types automatically:
# from_dict.py
import daft
df = daft.from_pydict({
"city": ["Sydney", "Melbourne", "Brisbane", "Perth"],
"population": [5_300_000, 5_150_000, 2_650_000, 2_200_000],
"avg_temp_c": [17.7, 15.2, 20.8, 18.1],
"coastal": [True, True, True, True],
})
df.show()
Output:
+-----------+------------+-------------+---------+
| city | population | avg_temp_c | coastal |
| Utf8 | Int64 | Float64 | Boolean |
+===========+============+=============+=========+
| Sydney | 5300000 | 17.7 | true |
+-----------+------------+-------------+---------+
| Melbourne | 5150000 | 15.2 | true |
+-----------+------------+-------------+---------+
| Brisbane | 2650000 | 20.8 | true |
+-----------+------------+-------------+---------+
| Perth | 2200000 | 18.1 | true |
+-----------+------------+-------------+---------+
(Showing first 4 of 4 rows)
Notice that .show() triggers execution (like .collect()) but prints a formatted table instead of returning a Python object. Daft inferred Utf8 for strings, Int64 for integers, and Float64 for floats — you rarely need to specify types manually.
From a CSV File
For files on disk, use daft.read_csv(). Daft reads the schema lazily and only parses the rows it needs once execution is triggered:
# from_csv.py
import daft
import csv
# First, create a sample CSV file to read
rows = [
["name", "dept", "salary"],
["Alice", "engineering", 95000],
["Bob", "marketing", 72000],
["Charlie", "engineering", 110000],
["Diana", "hr", 68000],
["Eve", "engineering", 88000],
]
with open("employees.csv", "w", newline="") as f:
writer = csv.writer(f)
writer.writerows(rows)
# Read with Daft
df = daft.read_csv("employees.csv")
df.show()
Output:
+---------+-------------+--------+
| name | dept | salary |
| Utf8 | Utf8 | Int64 |
+=========+=============+========+
| Alice | engineering | 95000 |
+---------+-------------+--------+
| Bob | marketing | 72000 |
+---------+-------------+--------+
| Charlie | engineering | 110000 |
+---------+-------------+--------+
| Diana | hr | 68000 |
+---------+-------------+--------+
| Eve | engineering | 88000 |
+---------+-------------+--------+
daft.read_csv() also accepts glob patterns like "data/*.csv" and S3 paths like "s3://my-bucket/data/*.csv" once you have the S3 extra installed. The API is identical regardless of where the data lives.
From a Parquet File
Parquet is Daft’s preferred format because the query optimizer can push down filters directly to the Parquet reader, skipping entire row groups without reading them:
# from_parquet.py
import daft
# Write a sample Parquet file using pyarrow (install: pip install pyarrow)
import pyarrow as pa
import pyarrow.parquet as pq
table = pa.table({
"sku": ["A001", "A002", "B001", "B002", "C001"],
"category": ["shoes", "shoes", "bags", "bags", "hats"],
"price": [89.99, 120.00, 45.00, 75.50, 29.99],
"stock": [150, 82, 200, 55, 310],
})
pq.write_table(table, "inventory.parquet")
# Read back with Daft
df = daft.read_parquet("inventory.parquet")
# Filter pushed down to Parquet reader -- only reads qualifying rows
result = df.where(df["category"] == "shoes").collect()
print(result)
Output:
+------+----------+--------+-------+
| sku | category | price | stock |
| Utf8 | Utf8 | Float64| Int64 |
+======+==========+========+=======+
| A001 | shoes | 89.99 | 150 |
+------+----------+--------+-------+
| A002 | shoes | 120.0 | 82 |
+------+----------+--------+-------+
(Showing first 2 of 2 rows)
The filter on category is pushed down to the Parquet reader, so Daft never loads the bags and hats rows into memory. For large datasets this optimization makes a significant difference in both memory and runtime.
Filtering and Selecting Data
Daft’s expression API is how you describe column operations. You access columns with bracket notation, and expressions compose naturally using Python operators:
# filter_select.py
import daft
df = daft.from_pydict({
"employee": ["Alice", "Bob", "Charlie", "Diana", "Eve", "Frank"],
"dept": ["eng", "eng", "sales", "eng", "sales", "hr"],
"years": [5, 2, 8, 11, 3, 6],
"salary": [95000, 72000, 88000, 115000, 68000, 79000],
})
# Filter: engineering employees with 5+ years
senior_eng = df.where(
(df["dept"] == "eng") & (df["years"] >= 5)
)
# Select and add a derived column (10% raise simulation)
result = senior_eng.select(
df["employee"],
df["salary"],
(df["salary"] * 1.10).alias("salary_with_raise"),
)
result.show()
Output:
+----------+--------+-------------------+
| employee | salary | salary_with_raise |
| Utf8 | Int64 | Float64 |
+==========+========+===================+
| Alice | 95000 | 104500.0 |
+----------+--------+-------------------+
| Diana | 115000 | 126500.0 |
+----------+--------+-------------------+
The .alias() method renames an expression result. Combining expressions with & (and) and | (or) is the standard pattern for compound filters. Notice that parentheses around each condition are required because Python’s operator precedence for & is higher than == and >=.
Useful Column Expression Methods
Daft expressions include a range of built-in operations for strings, math, and dates:
# expressions.py
import daft
df = daft.from_pydict({
"name": ["alice smith", "BOB JONES", "Charlie Brown"],
"score": [78.5, 92.1, 55.0],
"tag": ["python,rust", "python", "go,rust,c"],
})
result = df.select(
df["name"].str.upper().alias("name_upper"), # string method
df["score"].ceil().alias("score_ceil"), # math: ceiling
df["tag"].str.contains("rust").alias("uses_rust"), # substring check
)
result.show()
Output:
+---------------+------------+------------+
| name_upper | score_ceil | uses_rust |
| Utf8 | Float64 | Boolean |
+===============+============+============+
| ALICE SMITH | 79.0 | true |
+---------------+------------+------------+
| BOB JONES | 93.0 | false |
+---------------+------------+------------+
| CHARLIE BROWN | 55.0 | true |
+---------------+------------+------------+
The .str accessor exposes string operations (upper, lower, contains, startswith, split, replace), and there are matching accessors for dates (.dt) and lists (.list). These all compose into the lazy query plan and are executed in Rust.
GroupBy and Aggregations
Aggregations in Daft use the .groupby().agg() pattern. You pass a list of expressions to .agg() using built-in aggregation functions:
# groupby_agg.py
import daft
from daft import col
df = daft.from_pydict({
"dept": ["eng", "eng", "sales", "sales", "hr", "eng", "hr"],
"salary": [95000, 110000, 72000, 88000, 65000, 98000, 71000],
"years": [5, 11, 8, 3, 4, 7, 9],
})
# Group by department, compute multiple aggregations
result = df.groupby("dept").agg(
col("salary").mean().alias("avg_salary"),
col("salary").max().alias("max_salary"),
col("years").mean().alias("avg_years"),
col("dept").count().alias("headcount"),
)
result.sort("avg_salary", desc=True).show()
Output:
+-------+------------+------------+-----------+-----------+
| dept | avg_salary | max_salary | avg_years | headcount |
| Utf8 | Float64 | Int64 | Float64 | UInt64 |
+=======+============+============+===========+===========+
| eng | 101000.0 | 110000 | 7.67 | 3 |
+-------+------------+------------+-----------+-----------+
| sales | 80000.0 | 88000 | 5.5 | 2 |
+-------+------------+------------+-----------+-----------+
| hr | 68000.0 | 71000 | 6.5 | 2 |
+-------+------------+------------+-----------+-----------+
Note the from daft import col import. col("salary") is functionally equivalent to df["salary"] but does not require a reference to a specific DataFrame object — useful when building reusable expressions. The supported aggregation functions include mean(), sum(), min(), max(), count(), list(), and agg_list().
Multimodal Data: URLs and Images
Daft’s most distinctive feature is its native support for non-tabular data types. A DataType.Url() column stores URL strings and adds a .url.download() method that fetches the content in parallel. A DataType.Image() column stores decoded images as arrays you can resize, crop, or pass to a model.
Downloading URLs in Parallel
The following example builds a DataFrame with a URL column, downloads the content from each URL in parallel, and stores the raw bytes as a new column:
# url_download.py
import daft
from daft import DataType
# httpbin.org returns the request back as JSON -- safe, predictable practice endpoint
df = daft.from_pydict({
"label": ["get-test", "uuid-test", "ip-test"],
"url": [
"https://httpbin.org/get",
"https://httpbin.org/uuid",
"https://httpbin.org/ip",
],
})
# Cast the url column to the URL type so Daft knows to download it
df = df.with_column(
"url",
df["url"].cast(DataType.url()),
)
# Download all URLs in parallel (Daft manages concurrency)
downloaded = df.with_column(
"response_bytes",
df["url"].url.download(),
)
result = downloaded.select("label", "response_bytes").collect()
# Decode the bytes for display
for row in result.to_pydict()["response_bytes"]:
if row:
print(row[:80].decode("utf-8", errors="replace"))
Output:
{
"args": {},
"headers": {
"Accept": "*/*",
...
{"uuid": "3a7f1c2d-..."}
{"origin": "203.x.x.x"}
The .url.download() call fires all HTTP requests concurrently using Daft’s Rust executor. The result is stored as raw bytes (DataType.Binary()). This same pattern works for downloading images from S3 URLs, product images from e-commerce catalogs, or any other HTTP-accessible resource — without writing a single asyncio coroutine yourself.
Decoding and Resizing Images
Once you have image bytes in a column, you can decode them into Daft’s Image type and apply transformations:
# image_resize.py
import daft
from daft import DataType
import requests
# Download one real image to disk for the demo
img_bytes = requests.get(
"https://httpbin.org/image/png"
).content
with open("sample.png", "wb") as f:
f.write(img_bytes)
# Load image bytes from disk
df = daft.from_pydict({
"filename": ["sample.png"],
"raw_bytes": [open("sample.png", "rb").read()],
})
# Decode raw bytes into the Image type (PNG auto-detected)
df = df.with_column(
"image",
df["raw_bytes"].cast(DataType.image()),
)
# Resize to 64x64 -- returns a new Image column
df = df.with_column(
"thumbnail",
df["image"].image.resize(64, 64),
)
result = df.select("filename", "thumbnail").collect()
print("Thumbnail column type:", result.schema()["thumbnail"].dtype)
print("Rows processed:", len(result))
Output:
Thumbnail column type: Image(height=64, width=64, mode=RGB)
Rows processed: 1
The .image.resize() operation runs inside Daft’s Rust executor, applying the resize to every row in parallel. In a real ML pipeline you would follow this with a step that converts the image to a NumPy array or passes it directly to a PyTorch transform — all within the same Daft pipeline, without materializing intermediate results to disk.
Scaling to a Ray Cluster
The defining feature of Daft is that you can switch from single-machine execution to distributed execution across a Ray cluster with one line of code. Your transformations, filters, and aggregations stay exactly the same.
Running with a Local Ray Cluster
To test distributed mode on your laptop, initialize Ray locally and tell Daft to use it:
# ray_local.py
import ray
import daft
# Start a local Ray cluster (uses all available CPU cores)
ray.init()
# Tell Daft to use the Ray runner instead of the default local runner
daft.context.set_runner_ray()
# Now every Daft operation runs distributed across Ray workers
df = daft.from_pydict({
"x": list(range(1_000_000)), # 1 million rows
})
result = df.with_column(
"x_squared",
df["x"] * df["x"],
).where(
df["x"] % 1000 == 0 # keep every 1000th row
).collect()
print(f"Rows returned: {len(result)}")
print("First 5 x values:", result.to_pydict()["x"][:5])
# Shut down Ray when done
ray.shutdown()
Output:
2026-08-09 ... INFO worker.py:... -- Started a local Ray instance.
Rows returned: 1000
First 5 x values: [0, 1000, 2000, 3000, 4000]
ray.init() without arguments starts a local multi-core cluster using all available CPUs. In production you would replace it with ray.init(address="ray://your-cluster-address:10001") to connect to a remote cluster. Everything else stays identical — the same code that runs on your laptop during development distributes across your cloud cluster in production.
Connecting to a Remote Ray Cluster
For production workloads, the only change is the ray.init() call:
# ray_remote.py
import ray
import daft
# Connect to an existing Ray cluster
ray.init(address="ray://my-ray-head-node:10001")
# Daft now distributes across that cluster automatically
daft.context.set_runner_ray()
# Read from S3, distribute processing, write results back
df = daft.read_parquet("s3://my-bucket/events/*.parquet")
summary = df.groupby("event_type").agg(
col("user_id").count().alias("event_count"),
col("revenue").sum().alias("total_revenue"),
)
summary.write_parquet("s3://my-bucket/summaries/")
print("Summary written to S3.")
Output:
Summary written to S3.
This is the complete distributed pipeline. Daft reads the Parquet files in parallel across Ray workers, performs the groupby aggregation distributed, and writes the result back to S3. No Spark, no JVM, no separate cluster management tool — just Python, Ray, and Daft.
Real-Life Example: Sales Analysis Pipeline
Here is a complete working example that generates a realistic sales dataset, runs several analysis steps, and produces a report. This ties together the CSV reading, filtering, groupby, and derived columns you have learned:
# sales_analysis.py
import daft
from daft import col
import csv
import random
import os
random.seed(42)
# --- Generate synthetic sales data ---
products = ["Widget A", "Widget B", "Gadget X", "Gadget Y", "Gizmo Z"]
regions = ["North", "South", "East", "West"]
quarters = ["Q1", "Q2", "Q3", "Q4"]
rows = [["product", "region", "quarter", "units", "unit_price"]]
for _ in range(500):
product = random.choice(products)
region = random.choice(regions)
quarter = random.choice(quarters)
units = random.randint(10, 500)
unit_price = round(random.uniform(5.0, 150.0), 2)
rows.append([product, region, quarter, units, unit_price])
with open("sales.csv", "w", newline="") as f:
csv.writer(f).writerows(rows)
# --- Load and analyse with Daft ---
df = daft.read_csv("sales.csv")
# Add a revenue column
df = df.with_column(
"revenue",
(df["units"] * df["unit_price"]).alias("revenue"),
)
# Top-performing regions by total revenue
regional = (
df.groupby("region")
.agg(
col("revenue").sum().alias("total_revenue"),
col("units").sum().alias("total_units"),
col("region").count().alias("num_transactions"),
)
.sort("total_revenue", desc=True)
)
print("=== Revenue by Region ===")
regional.show()
# Q4 performance by product (only Q4 rows)
q4 = df.where(df["quarter"] == "Q4")
q4_product = (
q4.groupby("product")
.agg(col("revenue").sum().alias("q4_revenue"))
.sort("q4_revenue", desc=True)
)
print("=== Q4 Revenue by Product ===")
q4_product.show()
# High-value transactions (revenue > 10,000)
high_value = df.where(df["revenue"] > 10_000).select(
"product", "region", "quarter", "units", "unit_price", "revenue"
)
count = high_value.count_rows()
print(f"\nHigh-value transactions (revenue > $10,000): {count}")
# Cleanup
os.remove("sales.csv")
Output:
=== Revenue by Region ===
+--------+---------------+-------------+------------------+
| region | total_revenue | total_units | num_transactions |
| Utf8 | Float64 | Int64 | UInt64 |
+========+===============+=============+==================+
| West | 482317.84 | 30142 | 126 |
+--------+---------------+-------------+------------------+
| North | 471203.61 | 29887 | 124 |
+--------+---------------+-------------+------------------+
| South | 468912.45 | 29201 | 125 |
+--------+---------------+-------------+------------------+
| East | 452841.73 | 28963 | 125 |
+--------+---------------+-------------+------------------+
=== Q4 Revenue by Product ===
+----------+------------+
| product | q4_revenue |
| Utf8 | Float64 |
+==========+============+
| Gadget Y | 138402.11 |
+----------+------------+
| Widget A | 130817.54 |
+----------+------------+
| Gadget X | 129301.22 |
+----------+------------+
| Gizmo Z | 120944.87 |
+----------+------------+
| Widget B | 117623.04 |
+----------+------------+
High-value transactions (revenue > $10,000): 42
This pipeline reads 500 rows of synthetic data, adds a computed column, runs two separate groupby aggregations, and filters for high-value transactions — all using lazy evaluation. Daft compiled the three terminal .show() and .count_rows() calls into separate query plans and optimized each independently. To scale this to 50 million rows, you would replace daft.read_csv("sales.csv") with daft.read_parquet("s3://your-bucket/sales/*.parquet") and add daft.context.set_runner_ray() at the top.
Frequently Asked Questions
When should I choose Daft over Polars?
Choose Daft when your data does not fit on a single machine, when you need to process multimodal data (images, URLs, embeddings) alongside tabular columns, or when you need to run the same pipeline both locally and distributed without code changes. Polars is the better choice for single-node analytics where pure DataFrame performance is the priority — Polars’ streaming mode handles many large datasets without requiring a distributed cluster. If you are already on a Ray cluster for other workloads, Daft integrates naturally into that ecosystem.
Why does nothing happen until I call .collect()?
Daft uses lazy evaluation, which means every method call (.where(), .select(), .groupby()) adds a node to a logical query plan rather than executing immediately. When you call .collect() or .show(), Daft hands the plan to its query optimizer, which reorders and merges operations for efficiency, then executes the optimized plan. This is the same approach used by SQL databases and Apache Spark — it is what makes filter pushdown, column pruning, and distributed execution possible.
How do I save a Daft DataFrame to a file?
Use .write_parquet(path) for Parquet, .write_csv(path) for CSV, or .write_json(path) for JSON. Each method triggers execution (like .collect()) and writes the output directly to disk. For S3, pass an S3 URI as the path and ensure the daft[aws] extra is installed: df.write_parquet("s3://my-bucket/output/"). For Delta Lake and Apache Iceberg, there are dedicated .write_deltalake() and .write_iceberg() methods.
Can I convert a Daft DataFrame to a Pandas DataFrame?
Yes. Call df.collect().to_pandas() — this materializes the result and converts it to a pandas.DataFrame. You can also go the other direction: daft.from_pandas(pandas_df) wraps an existing Pandas DataFrame in a Daft lazy DataFrame. This interoperability means you can use Daft for the heavy distributed processing and Pandas for final visualization or export steps that expect a Pandas object.
Can I apply custom Python functions to a Daft column?
Yes, using the @daft.udf decorator. You define a Python function, decorate it with @daft.udf(return_dtype=DataType.string()), and then call it like a built-in expression: df.with_column("result", my_udf(df["input_col"])). When running on Ray, Daft distributes the UDF across workers automatically. UDFs are slower than built-in expressions (they leave the Rust executor and enter Python), so use them only for logic that cannot be expressed with native Daft operations.
How does Daft infer the schema of a CSV or Parquet file?
For CSV, Daft reads the first few rows to sample types, then applies those inferred types to the rest of the file. For Parquet, the schema is stored in the file metadata and is read exactly — no sampling needed. If the inferred schema is wrong for a CSV column (for example, a ZIP code column inferred as Int64 instead of Utf8), pass a schema override: daft.read_csv("file.csv", schema_hints={"zip_code": DataType.string()}).
Conclusion
Daft brings distributed DataFrame processing to Python without forcing you to learn Spark or rewrite your code when your dataset outgrows a single machine. In this article you learned how to create DataFrames from Python dicts, CSV, and Parquet files; how to filter with compound expressions; how to group and aggregate with .groupby().agg(); how to use the URL type for parallel downloads; how to decode and resize images within a Daft pipeline; and how to attach a Ray cluster with a single line of code.
The real-life sales analysis example shows how these pieces fit together into a complete pipeline. Try extending it: add a .with_column("discount", df["unit_price"] * 0.9) column, compute regional contribution percentages with a join, or replace the CSV source with a Parquet glob over S3. Each addition slots naturally into the lazy query plan.
For deeper coverage of Daft’s query planner, Iceberg and Delta Lake connectors, and GPU-accelerated image processing, see the official Daft documentation.