Daft Documentation

Technical Documentation: Eventual-Inc/Daft

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`benchmarking/ai/README.md`) Copy Section

AI Benchmarks

This repository contains performance benchmarks comparing different data processing engines (Daft, Ray Data, and Spark) across various AI and multimodal data processing workloads.

Overview Copy Chapter

The benchmarks cover four different workload types, each designed to test different aspects of multimodal data processing:

1. Audio Transcription - Transcribing audio files
2. Document Embedding - Generating embeddings for PDF documents
3. Image Classification - Classify images
4. Video Object Detection - Detect objects in videos

Performance Results Summary Copy Chapter

Overview (`benchmarking/asof_join/README.md`) Copy Section

Summary Copy Chapter

Adds a self-contained benchmarking suite for Daft's join_asof operation:

• data_generation.py — generates reproducible left/right parquet datasets at three scales (small, medium, large) with clustered timestamps and Zipf-skewed entity distribution, written to benchmarking/data/asof_join/
• benchmark.py — runs a single asof-join using Daft's native or Ray runner, wrapped in a memray memory tracker, and prints a JSON result with wall time and memray output path
• cluster.yaml — Ray cluster config for AWS (1 m7i.large head + 4 r7i.4xlarge workers) for distributed runs against S3

How to run Copy Chapter

From inside benchmarking/asof_join/:

1. Generate data (one-time)

python data_generation.py --scale small
# or --scale medium / --scale large / --all

2. Run locally (native runner)

python benchmark.py --scale small
# Output: asof_join_memray.bin + JSON result on stdout

3. Inspect memory profile

memray flamegraph asof_join_memray.bin

Run on a Ray cluster

Before running: update DATA_ROOT to your S3 bucket and uncomment daft.set_runner_ray() in benchmark.py. Also update the S3 bucket and IAM settings in cluster.yaml.

Spin up the cluster:

ray up benchmarking/asof_join/cluster.yaml

Forward the dashboard in one terminal:

ray dashboard benchmarking/asof_join/cluster.yaml

Submit the job in another (after updating DATA_ROOT and daft.set_runner_ray() in benchmark.py):

ray job submit --address "http://localhost:8265" --working-dir benchmarking/asof_join -- python benchmark.py --scale small

Tear down when done:

ray down benchmarking/asof_join/cluster.yaml

Overview (`benchmarking/lerobot/README.md`) Copy Section

LeRobot video decode: per-frame → per-shard

The daft.datasets.lerobot reader decoded video frames with a per-row UDF that
re-opened the MP4 shard for every frame. Because av.open() on a remote shard
re-reads and parses the container index over the network, decoding N frames
re-opened the shard N times, paying that cost each time - so cost scaled ~linearly
at ~3s/frame (the slope of the sweep below).

This directory holds the benchmarks that diagnosed it and the fix that makes the
decode batched: rows sharing a shard within a batch are grouped so each shard is
opened once per batch, instead of once per frame.

The fix: batched decode Copy Chapter

_decode_lerobot_video_timestamp in daft/datasets/lerobot.py is now a
@daft.func.batch UDF. Within each batch it groups rows by shard path, opens each
shard once, and does a single forward decode assigning the closest frame to every
requested timestamp. Output is byte-identical to the old per-row decode.

Original vs batched (rows 1→10) Copy Section

sweep.py times an end-to-end lerobot.read(...).limit(n).collect() with
frame decoding for n = 1..10 rows of a remote test dataset (pepijn223/egodex-test),
run once per reader revision (merge-base vs this branch) - the original grows linearly
to ~34s; the batched version stays flat at ~4s (all 10 frames share one shard → one
open). All benchmarks in this directory were run on the same machine, an Apple
M4 Max (36 GB).

8-frame output hashes matched exactly (sha 80bdb30c…) between versions.

Beyond this test dataset, the fix was validated on six public LeRobot v3 datasets
spanning av1/h264/mp4v, 5-30 fps, 128x128-1280x720, and 1-3 cameras - pixel-identical
output everywhere, 4-13x faster - plus a full-dataset decode and a 100-frame
comparison. See real_datasets.md.

Downstream workload: hand tracking Copy Section

hand_tracking.py times the
daft-physical-ai demo
workload end to end: decode 12 remote frames of pepijn223/egodex-test and run
MediaPipe hand tracking as a Daft UDF. run_hand_tracking.sh
runs it once per reader revision (the original pinned to the parent of the #7184
merge commit) and checks the detected hands match. Measured 2026-07-14 on an
Apple M4 Max (36 GB, macOS 15.6), remote reads over hf://:
44.8s → 9.8s, identical detections.

Running Copy Chapter

python sweep.py --label batched      # rows 1..10 sweep + chart
./run_hand_tracking.sh               # hand-tracking workload A/B + chart

Overview (`benchmarking/parquet/README.md`) Copy Section

Parquet Benchmarks

Goals:

1. Find Parquet features that Daft underperforms on
2. Compare Daft against other frameworks

Setup Copy Chapter

Create a new virtual environment and install the dependencies

python -m venv .venv
source .venv/bin/activate
pip install -r benchmark-requirements.txt

Now, install the version of Daft you wish to use for benchmarking (either a released wheel, or if you want, a local build)

pip install daft

Running the benchmarks: Copy Chapter

Goal 1: Find Parquet features that Daft underperforms on Copy Section

pytest benchmarking/parquet/ -m benchmark --benchmark-group-by=group -k daft

Goal 2: Compare Daft against other frameworks Copy Section

pytest benchmarking/parquet/ -m benchmark --benchmark-group-by=param:path

Check peak memory usage Copy Section

Ensure that you have pytest-memray installed.

pytest benchmarking/parquet/ -m benchmark --memray

Overview (`benchmarking/vllm/README.md`) Copy Section

vLLM Benchmarks

This directory contains benchmarks for running batch inference with vLLM.

Setup Copy Chapter

1. Install the dependencies:

• vllm
• daft
• ray

2. Run the generate_data.ipynb notebook to generate the datasets.
3. Update config.py with the correct dataset path and other parameters.
4. Run the benchmark scripts on a Ray cluster.

Benchmarks Copy Chapter

• naive-batch.py: Simple batch inference using a Daft batch function and vLLM's LLM class.
• naive-batch-sorted.py: Same as naive-batch.py, but with the prompts sorted.
• continuous-batch.py: Continuous batching using the vllm-prefix-caching provider, with prefix routing disabled.
• continuous-batch-sorted.py: Same as continuous-batch.py, but with the prompts sorted.
• prefix-bucketing.py: Both continuous batching and prefix bucketing using the vllm-prefix-caching provider.
• ray-data.py: Ray Data batching using ray.data.llm.build_llm_processor.

Overview (`docs/plugins/README.md`) Copy Section

Daft MkDocs Plugins

Custom MkDocs plugins for the Daft documentation site.

nav-hide-children Copy Chapter

A plugin that allows you to conditionally hide navigation children based on configured paths.

Configuration Copy Section

Add to your mkdocs.yml:

plugins:
  - nav-hide-children:
      hide_children:
        - ["Python API", "Functions"]  # Hide children of Python API > Functions
        - ["SQL Reference"]            # Hide children of SQL Reference

Usage Copy Section

The plugin automatically injects an attribute hide_children into navigation objects based on the config paths, which can be used in Jinja2 templates to conditionally hide navigation elements.

Overview (`examples/dvector/README.md`) Copy Section

dvector

Vector distance functions for Daft, implemented as a native extension.

Functions Copy Chapter

Vectors can be FixedSizeList, List, or LargeList columns. Float vectors support Float32 and Float64 elements.

Install Copy Chapter

Requires a Rust toolchain and setuptools-rust.

Usage Copy Chapter

import daft
import dvector
from dvector import l2_distance

# Load the extension into the active session
daft.load_extension(dvector)

df = daft.from_pydict({
    "a": [[1.0, 2.0, 3.0], [0.0, 0.0, 0.0]],
    "b": [[4.0, 5.0, 6.0], [1.0, 1.0, 1.0]],
})

df.select(l2_distance(daft.col("a"), daft.col("b"))).collect()
# ╭───────────╮
# │ result    │
# │ ---       │
# │ Float64   │
# ╞═══════════╡
# │ 5.196152  │
# ├╌╌╌╌╌╌╌╌╌╌╌┤
# │ 1.732051  │
# ╰───────────╯

Development Copy Chapter

# Build and install with the native library
uv pip install -e .

# Run tests
pytest -v tests/

Overview (`examples/hello/README.md`) Copy Section

Hello Native Extension

A minimal Daft native extension example. See the extension guide for details.

Quick start Copy Chapter

# Install the extension in the project .venv
uv pip install -e .

# Run the tests!
pytest -v

Overview (`examples/hello_cpp/README.md`) Copy Section

Hello C++ Native Extension

A minimal Daft native extension written in pure C++ using the Apache Arrow C++ library.

This is the C++ counterpart of the hello Rust example. It implements a single greet_cpp scalar function that prepends "Hello, " to every string in a column.

Prerequisites Copy Chapter

• CMake >= 3.20
• A C++17 compiler
• Apache Arrow C++ (with headers)

Install Arrow C++ Copy Section

macOS (Homebrew):

brew install apache-arrow

Ubuntu / Debian:

sudo apt install -y libarrow-dev

Conda:

conda install -c conda-forge arrow-cpp

Development Copy Chapter

Build the shared library Copy Section

cmake -B build -DCMAKE_EXPORT_COMPILE_COMMANDS=ON
cmake --build build

The CMAKE_EXPORT_COMPILE_COMMANDS flag generates build/compile_commands.json for clangd / LSP support. Symlink it to the project root if your editor needs it:

ln -sf build/compile_commands.json .

Install as a Python package Copy Section

uv pip install -e .

This uses scikit-build-core to build the C++ library and install it into the Python package.

Run tests Copy Section

pytest -v tests/

Usage Copy Chapter

import daft
import hello_cpp
from hello_cpp import greet
from daft import col
from daft.session import Session

sess = Session()
sess.load_extension(hello_cpp)

df = daft.from_pydict({"name": ["John", "Paul", "George", None]})

with sess:
    df.select(greet(col("name"))).show()

# ╭──────────────────╮
# │ greet_cpp         │
# │ ---               │
# │ Utf8              │
# ╞══════════════════╡
# │ Hello, John!      │
# ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
# │ Hello, Paul!      │
# ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
# │ Hello, George!    │
# ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
# │ None              │
# ╰──────────────────╯

Overview (`src/daft-dashboard/README.md`) Copy Section

daft-dashboard

Requirements Copy Chapter

• Node.js ≥22
• npm ≥11.10.0 — required for the min-release-age supply-chain setting in frontend/.npmrc to take effect. Older npm versions silently ignore it, leaving local npm install unprotected. Upgrade with npm install npm@latest.

Build Copy Chapter

The frontend (Next.js, under frontend/) is statically embedded into the Rust binary at compile time via include_dir! (see src/assets.rs). There is no separate dev server — any frontend change requires make build (or make build-release) from the repo root to take effect.

Architecture Copy Chapter

• Backend: Axum server (src/) exposes SSE streams and REST endpoints under /client/.
• Frontend: Next.js app (frontend/) compiled to static assets, embedded in the binary.
• State: DashboardState (in src/state.rs) holds query info in a DashMap (in memory, currently no persistence).

Overview (`src/daft-ext-macros/README.md`) Copy Section

daft-ext-macros

Proc macros for Daft extensions.

Most users should depend on daft-ext instead.

License Copy Chapter

Apache-2.0

Overview (`src/daft-ext/README.md`) Copy Section

daft-ext

Extension SDK for building custom Daft extensions.

Daft is a distributed query engine for large-scale data processing. This crate provides the types, traits, and macros needed to build extensions that plug into Daft's execution engine.

Usage Copy Chapter

[dependencies]
daft-ext = "0.1"

License Copy Chapter

Apache-2.0

Overview (`src/daft-scan/README.md`) Copy Section

daft-scan

Defines the ScanOperator trait, ScanTask struct, and built-in operator implementations (GlobScanOperator, AnonymousScanOperator). Also owns scan-related primitives: Pushdowns, PartitionField, Sharder, and predicate rewriting for partition pruning.

Overview (`tutorials/flyte/README.md`) Copy Section

Using Daft with Flyte

Daft works well with the Flyte workflow engine and its Ray plugin.

Using from Flyte task Copy Chapter

To run Daft in a Flyte task, use Daft as if you were using Daft in a local machine!

Using Daft from a Ray Flyte task Copy Chapter

To run Daft in a Ray Flyte task and connect to a Ray cluster, use Daft as if ray.init() has already been called.

That is to say, run Daft with: daft.set_runner_ray().

See: app.py for an example.

Setup Copy Chapter

To run this locally, perform Flyte local setup as detailed here: https://github.com/jeevb/flyte-examples/tree/main/ray

Overview (`benchmarking/ai/audio_transcription/README.md`) Copy Section

Audio Transcription Benchmark

Transcribes 113,800 audio files using Whisper-tiny model. Processes audio resampling, feature extraction, and speech-to-text inference.

Input Dataset: Common Voice 17 (S3 parquet format)
Output Format: Parquet with transcriptions and metadata
Cluster: 8 worker nodes using g6.xlarge instances
Benchmark Date: September 22, 2024
Framework Versions: Daft 0.6.2, Ray Data 2.49.2, AWS EMR Spark 7.10.0

Performance Results Copy Chapter

Overview (`benchmarking/ai/document_embedding/README.md`) Copy Section

Document Embedding Benchmark

Generates embeddings for 10,000 PDF documents. Extracts text, chunks content, and creates 384-dimensional embeddings using sentence-transformers/all-MiniLM-L6-v2 model across distributed nodes.

Input Dataset: Digital Corpora PDF metadata (S3 parquet format)
Output Format: Parquet with embeddings, text chunks, and metadata
Cluster: 8 worker nodes using g6.xlarge instances
Benchmark Date: September 22, 2024
Framework Versions: Daft 0.6.2, Ray Data 2.49.2, AWS EMR Spark 7.10.0

Performance Results Copy Chapter

Overview (`benchmarking/ai/image_classification/README.md`) Copy Section

Image Classification Benchmark

Classifies 803,580 rows (80,358 unique images, each repeated 10x) using ResNet18 model. Downloads images, applies preprocessing transforms, and runs inference to predict ImageNet labels across distributed GPU nodes.

Input Dataset: ImageNet benchmark dataset (S3 parquet format)
Output Format: Parquet with image URLs and predicted labels
Cluster: 8 worker nodes using g6.xlarge instances
Benchmark Date: September 22, 2024
Framework Versions: Daft 0.6.2, Ray Data 2.49.2, AWS EMR Spark 7.10.0

Performance Results Copy Chapter

Overview (`benchmarking/ai/video_object_detection/README.md`) Copy Section

Video Object Detection Benchmark

Detects objects in 1,000 videos using YOLO11n model. Extracts frames, runs object detection, and crops detected objects across distributed GPU nodes.

Input Dataset: Hollywood2 video dataset (S3 binary files)
Output Format: Parquet with object detections, bounding boxes, and cropped images
Cluster: 8 worker nodes using g6.xlarge instances
Benchmark Date: September 22, 2024
Framework Versions: Daft 0.6.2, Ray Data 2.49.2, AWS EMR Spark 7.10.0

Performance Results Copy Chapter

Overview (`k8s/charts/quickstart/README.md`) Copy Section

Daft Quickstart Helm Chart

A Helm chart for running Daft data processing workloads on Kubernetes, with support for both simple single-node execution and distributed Ray-based processing.

Features Copy Chapter

• Two deployment modes: Simple mode for quick jobs, distributed mode for large-scale processing
• Two execution methods: Script injection for rapid prototyping, custom images for managing complex dependencies
• No operator required: Uses native Kubernetes resources only

Prerequisites Copy Chapter

• Kubernetes 1.19+
• Helm 3.x
• kubectl configured to access your cluster

Quick Start Copy Chapter

Simple Mode (Default) Copy Section

Run a quick Daft job using the native runner:

# Create a simple Python script
cat > my_script.py <<'EOF'
# /// script
# dependencies = ["daft"]
# ///
import daft

df = daft.from_pydict({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35]
})

df.filter(df["age"] > 25).show()
EOF

# Deploy
helm install my-job oci://ghcr.io/eventual-inc/daft/quickstart \
  --set-file job.script=my_script.py

# View logs
kubectl logs -f job/my-job-quickstart-job

# Cleanup
helm uninstall my-job

Distributed Mode Copy Section

Run a Daft job on a Ray cluster with multiple workers:

# Create a simple Python script with Ray
cat > my_script.py <<'EOF'
# /// script
# dependencies = ["daft", "ray[client]==2.46.0"]
# ///

import daft
import ray

ray.init(runtime_env={"pip": ["daft"]})

df = daft.from_pydict({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35]
})

df.filter(df["age"] > 25).show()
EOF

# Deploy with Ray cluster
helm install distributed-job oci://ghcr.io/eventual-inc/daft/quickstart \
  --set distributed=true \
  --set worker.replicas=3 \
  --set-file job.script=my_script.py

# Access Ray Dashboard
kubectl port-forward service/distributed-job-quickstart-head 8265:8265
# Open http://localhost:8265

# Cleanup
helm uninstall distributed-job

Using a custom image Copy Section

This chart also supports running Daft jobs from custom images. This is useful in scenarios where more complex dependencies are required.

helm install custom-image-job oci://ghcr.io/eventual-inc/daft/quickstart \
  --set image=my-image:tag \
  --set "job.command={python,my_script.py}"

Installation Copy Chapter

See values.yaml for detailed descriptions of all supported parameters.

Monitoring Copy Chapter

View Job Status Copy Section

# Get job status
kubectl get jobs -l app.kubernetes.io/instance=my-release

# View job logs
kubectl logs -f job/my-release-quickstart-job

# Check pod status
kubectl get pods -l app.kubernetes.io/instance=my-release

Access Ray Dashboard (Distributed Mode) Copy Section

# Port forward to access both Ray Dashboard and Grafana
kubectl port-forward service/my-release-quickstart-head 8265:8265 3000:3000

# Open http://localhost:8265 in browser
# Grafana panels will be embedded directly in the Ray Dashboard

# Optionally open http://localhost:3000 to access Grafana directly
# Default credentials: admin/admin

Check Ray Cluster Status (Distributed Mode) Copy Section

# Get cluster status
kubectl exec deployment/my-release-quickstart-head -- ray status

# View cluster resources
kubectl exec deployment/my-release-quickstart-head -- python -c "import ray; ray.init(); print(ray.cluster_resources())"

Scaling Copy Chapter

Manual Scaling (Distributed Mode) Copy Section

# Scale workers using kubectl
kubectl scale deployment my-release-quickstart-worker --replicas=10

Troubleshooting Copy Chapter

Job Not Starting Copy Section

# Check job events
kubectl describe job my-release-quickstart-job

# Check pod events
kubectl describe pod -l app.kubernetes.io/component=job

# View pod logs
kubectl logs -l app.kubernetes.io/component=job

Workers Not Connecting (Distributed Mode) Copy Section

# Check head service
kubectl get service -l ray.io/node-type=head

# Check head logs
kubectl logs deployment/my-release-quickstart-head

# Check worker logs
kubectl logs deployment/my-release-quickstart-worker

Out of Memory Errors Copy Section

Increase memory limits in values:

job:
  resources:
    limits:
      memory: "16Gi"  # Increase as needed

head:
  resources:
    limits:
      memory: "32Gi"  # Increase as needed

worker:
  resources:
    limits:
      memory: "32Gi"  # Increase as needed

Script Not Found Error Copy Section

Ensure you're using --set-file (not --set):

# Correct
helm install my-job oci://ghcr.io/eventual-inc/daft/quickstart --set-file job.script=script.py

# Incorrect
helm install my-job oci://ghcr.io/eventual-inc/daft/quickstart --set job.script=script.py

Ray Module Not Found (Distributed Mode) Copy Section

Your custom image must include ray[default]:

# requirements.txt
daft
ray[default]

Uninstalling Copy Chapter

helm uninstall my-release

This removes all Kubernetes resources created by the chart.

License Copy Chapter

Apache 2.0

Contributing Copy Chapter

Contributions are welcome! Please see the Daft repository for contribution guidelines.

Support Copy Chapter

• Issues: GitHub Issues
• Documentation: docs.daft.ai
• Community: Daft Discussions

Overview (`tests/integration/gravitino/README.md`) Copy Section

Gravitino Integration Tests

These tests exercise Daft's Gravitino connector end-to-end. To run them:

1. Start the local Gravitino service:

   bash Copy

   cd tests/integration/gravitino/docker-compose
   docker compose up -d

2. Export the connection settings (if you use the compose file above the defaults already match):

   bash Copy

   export GRAVITINO_ENDPOINT=${GRAVITINO_ENDPOINT:-http://127.0.0.1:8090}
   export GRAVITINO_METALAKE=${GRAVITINO_METALAKE:-metalake_demo}

3. Run the tests:

   bash Copy

   DAFT_RUNNER=native pytest tests/integration/gravitino -v -m integration

Test Files Copy Chapter

• test_gravitino_fileset.py - Tests filesets with local file:// storage
• test_gravitino_fileset_s3.py - Tests filesets with S3 storage (using SeaweedFS)
• test_gravitino_table.py - Tests table catalog operations

S3 Storage Tests Copy Chapter

The S3 fileset tests (test_gravitino_fileset_s3.py) test Gravitino filesets backed by S3 storage.

Current Status Copy Section

The docker-compose setup includes:

1. SeaweedFS: S3-compatible storage on port 9000
2. Update fileset.conf: To run with S3, set a property in /root/gravitino/catalogs/fileset/conf/fileset.conf
3. Test Infrastructure: Complete test suite for S3 filesets

Running S3 Tests with External Gravitino Copy Section

If you have access to a Gravitino deployment with S3 filesets already configured, you can test against it:

# Point to an existing S3-backed fileset
export GRAVITINO_TEST_DIR="gvfs://fileset/<catalog>/<schema>/<fileset>/"
export GRAVITINO_TEST_FILE="gvfs://fileset/<catalog>/<schema>/<fileset>/file.parquet"
export GRAVITINO_ENDPOINT="http://your-gravitino-server:8090"

# Run specific tests
DAFT_RUNNER=native pytest tests/integration/gravitino/test_gravitino_fileset_s3.py::test_read_existing_s3_fileset -v -m integration

Test Coverage Copy Section

The S3 test suite includes:

• Basic read operations from S3-backed filesets
• Glob patterns and file listing
• Partitioned data handling
• Error handling for missing files and empty directories

Optional Configuration Copy Chapter

Optionally, configure GRAVITINO_TEST_FILE and GRAVITINO_TEST_DIR to point at a fileset that exists in
your Gravitino deployment so that the gvfs:// IO tests can read concrete data.

The catalog integration tests provision temporary catalog/schema/fileset via the Gravitino REST API and write
test data to either local storage or SeaweedFS. The tests clean up both the server-side metadata and storage when finished.

Overview (`tools/observability/opentelemetry/README.md`) Copy Section

OpenTelemetry

This directory contains the code for running the OpenTelemetry collector and exporter.

The docker compose file located in this directory will start the following services:

1. OpenTelemetry Collector - an agent for collecting and exporting opentelemetry data
2. Jaeger - an open source distributed tracing system
3. Prometheus - an open source metrics monitoring system

Running the collector Copy Chapter

To start the services, navigate to this directory and run the following command:

docker compose up

Executing a job with opentelemetry enabled Copy Chapter

The opentelemetry collector exposes an OTLP gRPC endpoint on port 4317.
To execute a job with opentelemetry enabled, set the DAFT_DEV_OTEL_EXPORTER_OTLP_ENDPOINT environment variable to grpc://localhost:4317 and run the job.

Example:

DAFT_DEV_OTEL_EXPORTER_OTLP_ENDPOINT=grpc://localhost:4317 python tools/observability/opentelemetry/example.py

Viewing the metrics Copy Chapter

Traces (Jaeger) Copy Section

To view the traces on the Jaeger UI, navigate to http://localhost:16686/ and select the daft service from the dropdown menu.

Metrics (Prometheus) Copy Section

To view the metrics on the Prometheus UI, navigate to http://localhost:9090/.

To query for all metrics provided by Daft, try the following PromQL query:

{service_name='daft'}

Index (`docs/index.md`) Copy Section

Introduction

Daft is a high-performance data engine designed for AI and multimodal workloads, providing simple, reliable data processing for images, audio, video, and structured data at any scale.

READ ANYTHING
      Daft reads raw unstructured and multimodal data collected from application systems
      Object storage (AWS S3, GCS, R2)Event bus (Kafka)Data lake (Iceberg, Deltalake)
    
    
      
    
  



  
    
      EXPENSIVE TRANSFORMATIONS
      Build efficient Daft data pipelines involving heavy transformations
      GPU modelsUser-provided Python codeExternal LLM APIs
    
    
      
    
  



  
    
      WRITE
      Daft lands data into specialized data systems for downstream use cases
      Search (full-text search and vector DBs)Applications (SQL/NoSQL databases)Analytics (data warehouses)Model training (S3 object storage)

Why Daft? Copy Chapter

:octicons-image-24: Unified multimodal data processing

While traditional dataframes struggle with anything beyond tables, Daft natively handles tables, images, text, and embeddings through a single Python API. No more stitching together specialized tools for different data types.

:material-language-python: Python-native, no JVM required

Built for modern AI/ML workflows with Python at its core and Rust under the hood. Skip the JVM complexity, version conflicts, and memory tuning to achieve 20x faster start times—get the performance without the Java tax.

:fontawesome-solid-laptop: Seamless scaling, from laptop to cluster

Start local, scale global—without changing a line of code. Daft's Rust-powered engine delivers blazing performance on a single machine and effortlessly extends to distributed clusters when you need more horsepower.

Key Features Copy Chapter

• Native multimodal processing: Process any data type—from structured tables to unstructured text and rich media—with native support for images, audio, video, and embeddings in a single, unified framework.

• Built-in AI operations: Transform data with AI natively: run LLM prompts with structured outputs, generate embeddings, and classify images or text using models from OpenAI, Transformers, or your own custom providers, all optimized for batch processing.

• Rust-powered performance: Experience breakthrough speed with our Rust foundation delivering vectorized execution and non-blocking I/O that processes the same queries with 5x less memory while consistently outperforming industry standards by an order of magnitude.

• Universal data connectivity: Access data anywhere it lives—cloud storage (S3, Azure, GCS, Hugging Face), modern table formats (Apache Iceberg, Delta Lake, Apache Hudi), or enterprise catalogs (Unity Catalog, AWS Glue)—all with zero configuration.

• Push your code to your data: Bring your Python functions directly to your data with zero-copy UDFs powered by Apache Arrow, eliminating data movement overhead and accelerating processing speeds.

• Out of the box reliability: Deploy with confidence—intelligent memory management prevents OOM errors while sensible defaults eliminate configuration headaches, letting you focus on results, not infrastructure.

!!! tip "Looking to get started with Daft ASAP?"

If you are ready to jump into code, take a look at these resources:

1. [Quickstart](quickstart.md): Itching to run some Daft code? Hit the ground running with our 10 minute quickstart.

2. [Examples](examples/index.md): See Daft in action with use cases across text, images, audio, and more.

3. [API Documentation](api/index.md): Searchable documentation and reference material to Daft's public API.

Contribute to Daft Copy Chapter

If you're interested in hands-on learning about Daft internals and would like to contribute to our project, join us on GitHub 🚀

Take a look at the many issues tagged with good first issue in our repo. If there are any that interest you, feel free to chime in on the issue itself or join us in our Distributed Data Slack Community and send us a message in #daft-dev. Daft team members will be happy to assign any issue to you and provide any guidance if needed!

Quickstart (`docs/quickstart.md`) Copy Section

Quickstart

Daft is the best multimodal data processing engine that allows you to load data from anywhere, transform it with a powerful DataFrame API and AI functions, and store it in your destination of choice. In this quickstart, you'll see what this looks like in practice with a realistic e-commerce data workflow.

Requirements Copy Section

Daft requires Python 3.10 or higher.

Install Daft Copy Section

You can install Daft using pip. Run the following command in your terminal or notebook:

pip install -U "daft[openai]"  # Includes OpenAI extras needed for this quickstart

Additionally, install these packages for image processing (used later in this quickstart):

pip install numpy pillow

Load Your Data Copy Section

Let's start by loading an e-commerce dataset from Hugging Face. This dataset contains 10,000+ Amazon products from diverse categories including electronics, toys, home goods, and more. Each product includes details like names, prices, descriptions, technical specifications, and product images.

import daft

df_original = daft.read_huggingface("calmgoose/amazon-product-data-2020")

!!! note "Load from anywhere"

Daft can load data from many sources including [S3](connectors/aws.md), [Iceberg](connectors/iceberg.md), [Delta Lake](connectors/delta_lake.md), [Hudi](connectors/hudi.md), and [more](connectors/index.md). We're using Hugging Face here as a demonstration.

Inspect Your Data Copy Section

Now let's take a look at what we loaded. You can inspect the DataFrame by simply printing it:

df_original

╭─────────┬──────────────┬──────────┬────────────┬──────────┬─────────────┬──────────────────╮
│ Uniq Id ┆ Product Name ┆ Category ┆      …     ┆ Variants ┆ Product Url ┆ Is Amazon Seller │
│ ---     ┆ ---          ┆ ---      ┆            ┆ ---      ┆ ---         ┆ ---              │
│ String  ┆ String       ┆ String   ┆ (9 hidden) ┆ String   ┆ String      ┆ String           │
╰─────────┴──────────────┴──────────┴────────────┴──────────┴─────────────┴──────────────────╯

(No data to display: Dataframe not materialized, use .collect() to materialize)

You see the above output because Daft is lazy by default - it displays the schema (column names and types) but doesn't actually load or process your data until you explicitly tell it to. This allows Daft to optimize your entire workflow before executing anything.

To actually view your data, you have two options:

Option 1: Preview with .show() - View the first few rows:

df_original.show(2)

/* Detailed source-code truncated for AI context efficiency. */

This materializes and displays just the first 2 rows, which is perfect for quickly inspecting your data without loading the entire dataset.

Option 2: Materialize with .collect() - Load the entire dataset:

# df_original.collect()

This would materialize the entire DataFrame (all 10,000+ rows in this case) into memory. Use .collect() when you need to work with the full dataset in memory.

Working with a Smaller Dataset Copy Section

For quick experimentation, let's create a smaller, simplified version of the dataframe with just the essential columns:

# Select only the columns we need and limit to 5 rows for faster iteration
df = df_original.select("Product Name", "About Product", "Image").limit(5)

Now we have a manageable dataset of 5 products with just the product name, description, and image URLs. This simplified dataset lets us explore Daft's features without the overhead of unnecessary columns.

Downloading Images Copy Section

Let's extract and download product images. The Image column contains pipe-separated URLs. We'll extract the first URL and download it:

# Extract the first image URL from the pipe-separated list
# The pattern captures everything before the first pipe or the entire string if no pipe
df = df.with_column(
    "first_image_url",
    daft.functions.regexp_extract(
        df["Image"],
        r"^([^|]+)",  # Extract everything before the first pipe
        1  # Get the first capture group
    )
)

# Download the image data
df = df.with_column(
    "image_data",
    daft.functions.download(df["first_image_url"], on_error="null")
)

# Decode images for visual display (in Jupyter notebooks, this shows actual images!)
df = df.with_column(
    "image",
    daft.functions.decode_image(df["image_data"], on_error="null")
)

# Check what we have - in Jupyter notebooks, the 'image' column shows actual images!
df.select("Product Name", "first_image_url", "image_data", "image").show(3)

╭────────────────────────────────┬────────────────────────────────┬────────────────────────────────┬──────────────╮
│ Product Name                   ┆ first_image_url                ┆ image_data                     ┆ image        │
│ ---                            ┆ ---                            ┆ ---                            ┆ ---          │
│ String                         ┆ String                         ┆ Binary                         ┆ Image[MIXED] │
╞════════════════════════════════╪════════════════════════════════╪════════════════════════════════╪══════════════╡
│ DB Longboards CoreFlex Crossb… ┆ https://images-na.ssl-images-… ┆ b"\xff\xd8\xff\xe0\x00\x10JFI… ┆ <Image>      │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ Electronic Snap Circuits Mini… ┆ https://images-na.ssl-images-… ┆ b"\xff\xd8\xff\xe0\x00\x10JFI… ┆ <Image>      │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ 3Doodler Create Flexy 3D Prin… ┆ https://images-na.ssl-images-… ┆ b"\xff\xd8\xff\xe0\x00\x10JFI… ┆ <Image>      │
╰────────────────────────────────┴────────────────────────────────┴────────────────────────────────┴──────────────╯

(Showing first 3 rows)

!!! note "Visual Display in Notebooks"

In Jupyter notebooks, the `image` column will display actual thumbnail images instead of `` text.

This demonstrates Daft's multimodal capabilities:

• Native regex support: Use regexp_extract() to parse structured text with Rust-powered regex
• URL handling: Download content directly withdaft.functions.download()
• Image decoding: Convert binary data to images with decode_image() for visual display

The decoded images are now ready for further processing.

Batch AI Inference on Images Copy Section

Let's use AI to analyze product materials at scale. Daft automatically parallelizes AI operations across your local machine's cores, making it efficient to process multiple images concurrently.

Let's suppose you want to create a new column that shows if each product is made of wood or not. This might be useful for, for example, a filtering feature on your website.

If you're running this in Google Colab or Jupyter, run the following cell to set your OpenAI API key. In Colab, first add your key to Secrets (🔑 icon in the left sidebar) with the name OPENAI_API_KEY. In Jupyter, you'll be prompted to enter your key and the input will be hidden.

import os
try:
    from google.colab import userdata
    os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
except ImportError:
    from getpass import getpass
    if "OPENAI_API_KEY" not in os.environ:
        os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API key: ")

from pydantic import BaseModel, Field
from daft.functions import prompt

# Define a simple structured output model
class WoodAnalysis(BaseModel):
    is_wooden: bool = Field(description="Whether the product appears to be made of wood")

# Run AI inference on each image - Daft automatically batches and parallelizes this
df = df.with_column(
    "wood_analysis",
    prompt(
        ["Is this product made of wood? Look at the material.", df["image"]],
        return_format=WoodAnalysis,
        model="gpt-4o-mini",  # Using mini for cost-efficiency
        provider="openai",
        # api_key="your-key-here",  # Use OPENAI_API_KEY env var, or uncomment to set manually
    )
)

# Extract the boolean value from the structured output
# The result is a struct, so we extract the 'is_wooden' field
df = df.with_column(
    "is_wooden",
    df["wood_analysis"]["is_wooden"]
)

# Materialize the dataframe to compute all transformations
df = df.collect()

# View results
df.select("Product Name", "image", "is_wooden").show()

╭────────────────────────────────┬──────────────┬───────────╮
│ Product Name                   ┆ image        ┆ is_wooden │
│ ---                            ┆ ---          ┆ ---       │
│ String                         ┆ Image[MIXED] ┆ Bool      │
╞════════════════════════════════╪══════════════╪═══════════╡
│ DB Longboards CoreFlex Crossb… ┆ <Image>      ┆ true      │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌┤
│ Electronic Snap Circuits Mini… ┆ <Image>      ┆ false     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌┤
│ Guillow Airplane Design Studio ┆ <Image>      ┆ false     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌┤
│ Woodstock- Collage 500 pc Puz… ┆ <Image>      ┆ false     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌┤
│ 3Doodler Create Flexy 3D Prin… ┆ <Image>      ┆ false     │
╰────────────────────────────────┴──────────────┴───────────╯

(Showing first 5 of 5 rows)

The AI analyzes each product image to determine if it's made of wood. Notice that the longboard is identified as wooden (true), while the electronic circuits, design studio, puzzle, and 3D printing filament are identified as not wooden (false).

!!! note "Improving Accuracy"

Looking at the actual product data, the longboard is made of bamboo and fiberglass, not wood. However, this is exactly what a human might categorize from the image alone! To improve accuracy, you could feed additional context to the AI like the product name, category, and description alongside the image. This example demonstrates how to get started with image-based analysis.

Expanding the Analysis Copy Section

Now, suppose you're satisfied with the results from your small subset and want to scale up. Instead of analyzing just 5 products, let's run the same analysis on 100 products to get more meaningful insights:

/* Detailed source-code truncated for AI context efficiency. */

Out of 100 products analyzed:
  - 4 are made of wood
  - 96 are not made of wood
  - Percentage of wooden products: 4.0%

!!! note "Results May Vary"

AI models are non-deterministic, so you may see slightly different numbers when running this analysis.

Storing Your Results Copy Section

After processing your data, you'll often want to save it for later use. Let's store our analyzed dataset as Parquet files:

# Write the analyzed data to local Parquet files
df_large.write_parquet("product_analysis", write_mode="overwrite")

This writes your data to the product_analysis/ directory. Daft automatically handles file naming using UUIDs to prevent conflicts. The write_mode="overwrite" parameter ensures that any existing data in the directory is replaced.

!!! note "Write anywhere"

Just like reading, Daft can write data to many destinations including [S3](connectors/aws.md), [Iceberg](connectors/iceberg.md), [Delta Lake](connectors/delta_lake.md), and [more](connectors/index.md).

Loading Your Stored Data Copy Section

Let's verify the stored data by loading it back from those Parquet files:

# Read the data back from Parquet files
df_loaded = daft.read_parquet("product_analysis/*.parquet")

# Verify the data loaded correctly
df_loaded.show(5)

/* Detailed source-code truncated for AI context efficiency. */

What's Next? Copy Section

Now that you have a basic sense of Daft's functionality and features, here are some more resources to help you get the most out of Daft:

!!! tip "Scaling Further"

This same pipeline can process thousands or millions of products by leveraging Daft's distributed computing capabilities. Check out our [distributed computing guide](distributed/index.md) to run this analysis at scale on Ray or Kubernetes clusters.

Work with your favorite table and catalog formats:

• Apache Hudi
• Apache Iceberg
• AWS Glue
• AWS S3 Tables
• Delta Lake
• Hugging Face Datasets
• Unity Catalog
• Apache Gravitino

Explore our Examples to see Daft in action:

• :material-image-edit: MNIST Digit Classification
• :octicons-search-16: Running LLMs on the Red Pajamas Dataset
• :material-image-search: Querying Images with UDFs
• :material-image-sync: Image Generation on GPUs
• :material-window-closed-variant: Window Functions in Daft

Ai Functions Overview (`docs/ai-functions/overview.md`) Copy Section

AI Functions Overview

Daft is purpose-built for scaling multimodal AI workloads. AI Functions make it easy to run models on data whether you're generating embeddings, classifying text, or prompting language models. Daft's built-in Image and File abstractions simplify data handling whether it's on your laptop or in the cloud.

The current list of AI functions includes:

• prompt - Generate text completions from language models
• embed_text - Create vector embeddings from text
• embed_image - Create vector embeddings from images
• classify_text - Zero-shot text classification
• classify_image - Zero-shot image classification

For more detailed information on the Providers, see AI Providers Overview. If you'd like to contribute a new AI function or expand provider support, check out Contributing New AI Functions & Providers.

See Also Copy Chapter

• Working with Text - Text processing and chunking strategies
• Working with Images - Image processing and transformations
• Embeddings - Vector representations for RAG and search
• Custom Code with @daft.func - Write custom Python functions for specialized processing
• Batch Inference - End-to-end batch inference patterns
• Checkpointing - Resume expensive AI pipelines without redoing completed work

Api Index (`docs/api/index.md`) Copy Section

Daft API Documentation

Welcome to Daft Python API Documentation. For Daft User Guide, head to User Guide.

• AI

  Functions and interfaces for building AI applications with daft. This includes providers
  and various model inference protocols such as text embedding.

• I/O

  Variety of approaches to creating a DataFrame from reading various data sources like in-memory data, files, data catalogs, and integrations and writing to various data sources.

• DataFrame

  Available DataFrame methods that are enqueued in the DataFrame's internal query plan and executed when Execution DataFrame methods are called.

• Datasets

  Simple, performant, and responsible ways to access useful datasets like Common Crawl and DROID.

• Expressions

  Expressions allow you to express some computation that needs to happen in a DataFrame.

• Functions

  Daft provides a set of built-in operations that can be applied to DataFrame columns.

• User-Defined Functions

  User-Defined Functions (UDFs) are a mechanism to run Python code on the data that lives in a DataFrame.

• Window

  Window functions allow you to perform calculations across a set of rows that are related to the current row.

• Sessions

  Sessions enable you to attach catalogs, tables, and create temporary objects which are accessible through both the Python and SQL APIs.

• Catalogs & Tables

  Daft integrates with various catalog implementations using its Catalog and Table interfaces to manage catalog objects like tables and namespaces.

• Schema

  Daft can display your DataFrame's schema without materializing it by performing intelligent sampling of your data to determine appropriate schema.

• Data Types

  Daft provides simple DataTypes that are ubiquituous in many DataFrames such as numbers, strings, dates, tensors, and images.

• Aggregations

  When performing aggregations such as sum, mean and count, Daft enables you to group data by certain keys and aggregate within those keys.

• Series

  Series expose methods which invoke high-performance kernels for manipulation of a column of data.

• Configuration

  Configure the execution backend, Daft in various ways during execution, and how Daft interacts with storage.

• Miscellaneous

Architecture Index (`docs/architecture/index.md`) Copy Section

Daft Architecture

Daft is a high-performance data engine designed for any modality and any scale. At a high level, a Daft query moves through 3 main stages: Planning, Optimization, and Execution.

Planning Copy Chapter

Users express their workloads via the Python DataFrame API or the SQL interface. These APIs are lazy: operations construct a plan of execution, but they do not read or transform data yet. Internally, the workload is represented as a LogicalPlan: a tree of operators that describe what work should happen. Example operators include Source, Project, Filter, GroupBy, and Join.

Within each operator, especially the Project operator, there is a collection of expressions that define how data should be transformed. When users apply native functions (like col("a") + 1) or custom Python UDFs (such as col("b").apply(my_udf)), these are represented internally as nodes in an expression tree. Each node in the tree can be a column reference, a literal, a function call, or a UDF, and complex expressions are built by chaining these nodes together.

For example, consider the following user code:

import daft
from daft import col

df = daft.from_pydict({"a": [1, 2, 3]})
result = df.select((col("a") + 1).alias("a_plus_one"))
result.explain()

Internally, this becomes a simple expression attached to the Project operator:

* Project: col(a) + lit(1) as a_plus_one
|
* Source:
|   Number of partitions = 1
|   Output schema = a#Int64

Optimization Copy Chapter

Executing the logical plan as written would be correct but often inefficient. Daft applies both rule-based and cost-based optimization once execution is requested.

• Rule-based pass. Classical rewrite rules execute first: filter, projection, limit, and aggregation pushdowns; projection folding and splitting; pruning redundant repartitions; expression simplification; and subquery unnesting. These transform the plan structure without needing runtime statistics.
• Cost-based pass. Joins are reordered using a brute-force enumerator that evaluates the cheapest ordering based on available statistics calculated from sources.
• Multimodal awareness. Expensive projections such as Python UDFs, model inference, URL downloads, image decoding, and similar operators are isolated into dedicated logical nodes. They are intentionally not pushed into scans so that execution can batch, schedule, and backpressure them independently. Such operations are executed as late as correctness permits (after joins, aggregations, etc.) to reduce wasted work on discarded rows.

This hybrid optimizer lets users focus on declarative queries while still obtaining efficient plans tailored to multimodal workloads.

Execution Copy Chapter

Execution is triggered by a materializing operation such as collect(), show(), to_arrow(), and write methods such as write_parquet().

Once optimization completes, the plan is executed by either the native runner (single-machine) or the Ray runner (distributed).

Native Runner (aka Swordfish) Copy Section

The native runner is a streaming execution engine implemented in Rust, using the Tokio runtime for async I/O and multithreaded operations.

Given a plan, the native runner constructs a graph of operators, where each operator corresponds to a node in the plan. Operators send data up the graph via async channels.
Once the graph is constructed, the runner initiates execution, which typically follows this pattern:

• Source operators read data from local files or object stores and emit them in batches.
• Intermediate operators (e.g. Project, Filter, UDF) receive batches of data from previous operators, immediately transform them, and send them to the next operator.
• Sinks receive and accumulate batches of data. Streaming sinks (e.g. Limit) can emit early, while blocking sinks (e.g. Aggregate, Sort) only emit results once all input has been processed.

Operators determine their own parallelism and batching, and the engine schedules work across operators on a threadpool (Tokio).

Ray Runner (aka Flotilla) Copy Section

The Ray runner is a distributed execution engine built on top of Ray.

A Ray cluster contains a single head node, and all other nodes are worker nodes. On each worker node, Daft launches a single Ray actor that hosts an instance of the Swordfish engine.

The Flotilla scheduler runs on the head node and distributes tasks across workers, monitors progress, and orchestrates data movement when needed. Each task corresponds to a partition of the DataFrame, and tasks are assigned based on data locality and worker load, so computation runs close to the data while keeping cluster resources balanced. The number of partitions typically corresponds to the number of input files.

!!! tip "Controlling partitioning"

See our [Partitioning and Batching guide](../optimization/partitioning.md) to learn how to control the number and distribution of partitions in your DataFrame

Each worker receives tasks and executes them using the Swordfish streaming engine with full access to the machine's resources. Outputs of the tasks are written to Ray's object store, and can be moved between workers for global operations like groupbys and joins.

Benchmarks Index (`docs/benchmarks/index.md`) Copy Section

Benchmarks

AI Benchmarks Copy Chapter

!!! info "Powered by Flotilla"

These benchmarks showcase [Flotilla](https://www.daft.ai/blog/introducing-flotilla-simplifying-multimodal-data-processing-at-scale), Daft's distributed engine optimized for multimodal data processing.

[Daft vs Spark vs Ray Data: Benchmarks for Multimodal AI Workloads](https://www.daft.ai/blog/benchmarks-for-multimodal-ai-workloads) - Full methodology and technical deep-dive.

Modern AI workloads demand simple, fast, and scalable data engines. Daft is purpose-built for these workloads, excelling where traditional big data frameworks fall short.

Our benchmarks show how Daft uniquely delivers on the core needs of AI data processing:

1. Process any modality: Audio, video, images, and documents as first class citizens.
2. Scale from 1 to 1000: Easily scale across machines without rewriting code or configs.
3. Run models over data: Seamlessly use models to embed, classify, and generate data.

Setup Copy Section

The AI benchmarks cover four different workload types, each designed to test different use cases across modalities, and at large scale. Each benchmark runs on AWS g6.xlarge instances with 8 worker nodes.

The complete benchmark code is available in our repository:

• Audio Transcription - Transcribing 113,800 audio files using OpenAI Whisper-tiny model
• Document Embedding - Generating embeddings for 10,000 PDF documents using sentence-transformers
• Image Classification - Classifying 803,580 images using ResNet18 model
• Video Object Detection - Detecting objects in 1,000 videos using YOLO11n model

Each benchmark includes implementations for Daft, Ray Data, and EMR Spark, along with cluster configurations and dependencies.

Benchmark Date: September 22, 2025

Engine Versions: Daft 0.6.2, Ray Data 2.49.2, AWS EMR Spark 7.10.0

Results Copy Section

<script type="text/javascript">window.PlotlyConfig = {MathJaxConfig: 'local'};</script>
<script charset="utf-8" src="https://cdn.plot.ly/plotly-2.20.0.min.js"></script>

<script type="text/javascript">
    window.PLOTLYENV = window.PLOTLYENV || {};
    if (document.getElementById("ai-benchmarks-chart")) {
        Plotly.newPlot(
            "ai-benchmarks-chart",
            [
                {
                    "marker": {"color": "rgba(255, 0, 255, 1)"},
                    "name": "Daft",
                    "x": ["Audio Transcription", "Document Embedding", "Image Classification", "Video Object Detection"],
                    "y": [6.37, 1.9, 4.38, 11.77],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["4.6x Slower", "7.6x Slower", "5.4x Slower", "2.2x Slower"],
                    "marker": {"color": "rgba(0, 102, 255, 0.75)"},
                    "name": "Ray Data",
                    "x": ["Audio Transcription", "Document Embedding", "Image Classification", "Video Object Detection"],
                    "y": [29.33, 14.53, 23.5, 25.9],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["4.0x Slower", "4.2x Slower", "10.3x Slower", "18.4x Slower"],
                    "marker": {"color": "rgba(226,90,28, 0.75)"},
                    "name": "Spark",
                    "x": ["Audio Transcription", "Document Embedding", "Image Classification", "Video Object Detection"],
                    "y": [25.77, 8.07, 45.12, 216.0],
                    "type": "bar",
                    "textposition": "inside"
                }
            ],
            {
                "title": {"text": "AI Benchmarks - Performance Comparison (lower is better)"},
                "yaxis": {
                    "title": {"text": "Time (minutes, log scale)"},
                    "type": "log",
                    "tickmode": "array",
                    "tickvals": [1, 2, 5, 10, 20, 40, 60, 220],
                    "ticktext": ["1", "2", "5", "10", "20", "40", "60", "220"]
                },
                "xaxis": {"title": {"text": "Workload"}},
                "uniformtext": {"minsize": 8, "mode": "hide"}
            },
            {"displayModeBar": false, "responsive": true}
        );
    }
</script>

Logs Copy Section

Complete execution logs for all AI benchmark runs are available for transparency and reproducibility:

TPC-H Benchmarks Copy Chapter

Here we compare Daft against some popular Distributed Dataframes such as Spark, Modin, and Dask on the TPC-H benchmark. Our goal for this benchmark is to demonstrate that Daft is able to meet the following development goals:

1. Solid out of the box performance: great performance without having to tune esoteric flags or configurations specific to this workload
2. Reliable out-of-core execution: highly performant and reliable processing on larger-than-memory datasets, without developer intervention and Out-Of-Memory (OOM) errors
3. Ease of use: getting up and running should be easy on cloud infrastructure for an individual developer or in an enterprise cloud setting

A great stress test for Daft is the TPC-H benchmark, which is a standard benchmark for analytical query engines. This benchmark helps ensure that while Daft makes it very easy to work with multimodal data, it can also do a great job at larger scales (terabytes) of more traditional tabular analytical workloads.

Setup Copy Section

The basic setup for our benchmarks are as follows:

1. We run questions 1 to 10 of the TPC-H benchmarks using Daft and other commonly used Python Distributed Dataframes.
2. The data for the queries are stored and retrieved from AWS S3 as partitioned Apache Parquet files, which is typical of enterprise workloads. No on disk/in-memory caching was performed.
3. We run each framework on a cluster of AWS i3.2xlarge instances that each have:
   • 8 vCPUs
   • 61G of memory
   • 1900G of NVMe SSD space

The frameworks that we benchmark against are Spark, Modin, and Dask. We chose these comparable Dataframes as they are the most commonly referenced frameworks for running large scale distributed analytical queries in Python.

For benchmarking against Spark, we use AWS EMR which is a hosted Spark service. For other benchmarks, we host our own Ray and Dask clusters on Kubernetes. Please refer to the section on our Detailed Benchmarking Setup for additional information.

Results Copy Section

!!! success "Highlights"

1. Out of all the benchmarked frameworks, **only Daft and EMR Spark are able to run terabyte scale queries reliably** on out-of-the-box configurations.
2. **Daft is consistently much faster** (3.3x faster than EMR Spark, 7.7x faster than Dask Dataframes, and 44.4x faster than Modin).

!!! note "Note"

We were unable to obtain full results for Modin due to cluster OOMs, errors and timeouts (one hour limit per question attempt). Similarly, Dask was unable to provide comparable results for the terabyte scale benchmark. It is possible that these frameworks may perform and function better with additional tuning and configuration. Logs for all the runs are provided in a public AWS S3 bucket.

100 Scale Factor

First we run TPC-H 100 Scale Factor (around 100GB) benchmark on 4 i3.2xlarge worker instances. In total, these instances add up to 244GB of cluster memory which will require the Dataframe library to perform disk spilling and out-of-core processing for certain questions that have a large join or sort.

<script type="text/javascript">window.PlotlyConfig = {MathJaxConfig: 'local'};</script>
<script charset="utf-8" src="https://cdn.plot.ly/plotly-2.20.0.min.js"></script>

<script type="text/javascript">
    window.PLOTLYENV = window.PLOTLYENV || {};
    if (document.getElementById("78330a19-a541-460b-bd9f-217b9d4cd137")) {
        Plotly.newPlot(
            "78330a19-a541-460b-bd9f-217b9d4cd137",
            [
                {
                    "marker": {"color": "rgba(255, 0, 255, 1)"},
                    "name": "Daft",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [1.0666666666666667, 0.7666666666666667, 0.9833333333333333, 1.05, 1.9666666666666666, 0.6333333333333333, 1.1666666666666667, 2.25, 2.183333333333333, 1.0166666666666666],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["5.6x Slower", "1.1x Slower", "5.1x Slower", "2.8x Slower", "2.0x Slower", "9.7x Slower", "4.3x Slower", "2.0x Slower", "2.3x Slower", "4.8x Slower"],
                    "marker": {"color": "rgba(226,90,28, 0.75)"},
                    "name": "Spark",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [5.991666666666666, 0.8716666666666666, 4.996666666666667, 2.955, 3.8583333333333334, 6.135000000000001, 4.985, 4.428333333333333, 5.051666666666667, 4.863333333333333],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["4.2x Slower", "1.4x Slower", "6.9x Slower", "13.0x Slower", "8.2x Slower", "6.1x Slower", "6.8x Slower", "3.6x Slower", "11.8x Slower", "12.1x Slower"],
                    "marker": {"color": "rgba(255,193,30, 0.75)"},
                    "name": "Dask",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [4.456666666666666, 1.0983333333333334, 6.748333333333333, 13.615, 16.215, 3.8366666666666664, 7.96, 8.148333333333333, 25.790000000000003, 12.306666666666667],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["29.1x Slower", "12.5x Slower", "nanx Slower", "48.6x Slower", "nanx Slower", "87.7x Slower", "nanx Slower", "nanx Slower", "nanx Slower", "52.7x Slower"],
                    "marker": {"color": "rgba(0,173,233, 0.75)"},
                    "name": "Modin",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [31.066666666666666, 9.616666666666667, null, 51.05, null, 55.53333333333333, null, null, null, 53.6],
                    "type": "bar",
                    "textposition": "inside"
                }
            ],
            {
                "title": {"text": "TPCH 100 Scale Factor - 4 Nodes (lower is better)"},
                "yaxis": {"title": {"text": "Time (minutes)"}},
                "xaxis": {"title": {"text": "TPCH Question"}},
                "uniformtext": {"minsize": 8, "mode": "hide"}
            },
            {"displayModeBar": false, "responsive": true}
        );
    }
</script>

* Only for queries that completed.

From the results we see that Daft, Spark, and Dask are able to complete all the questions and Modin completes less than half. We also see that Daft is 3.3x faster than Spark and 7.7x faster than Dask including S3 IO. We expect these speed-ups to be much larger if the data is loaded in memory instead of cloud storage, which we will show in future benchmarks.

1000 Scale Factor

Next we scale up the data size by 10x while keeping the cluster size the same. Since we only have 244GB of memory and 1TB+ of tabular data, the DataFrame library will be required to perform disk spilling and out-of-core processing for all questions at nearly all stages of the query.

<script type="text/javascript">window.PlotlyConfig = {MathJaxConfig: 'local'};</script>
<script charset="utf-8" src="https://cdn.plot.ly/plotly-2.20.0.min.js"></script>

<script type="text/javascript">
    window.PLOTLYENV = window.PLOTLYENV || {};
    if (document.getElementById("2e3c4bff-c808-4722-8664-d4c63ee41e55")) {
        Plotly.newPlot(
            "2e3c4bff-c808-4722-8664-d4c63ee41e55",
            [
                {
                    "marker": {"color": "rgba(255, 0, 255, 1)"},
                    "name": "Daft",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [4.85, 9.766666666666667, 12.933333333333334, 11.233333333333333, 17.616666666666667, 2.7, 15.15, 18.5, 22.833333333333332, 13.983333333333333],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["12.1x Slower", "0.9x Slower", "3.8x Slower", "2.9x Slower", "2.1x Slower", "22.3x Slower", "3.5x Slower", "2.7x Slower", "2.6x Slower", "3.4x Slower"],
                    "marker": {"color": "rgba(226,90,28, 0.75)"},
                    "name": "Spark",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [58.625, 8.591666666666667, 48.559999999999995, 32.88666666666667, 36.98166666666667, 60.11333333333334, 52.34, 49.475, 58.26166666666666, 46.85333333333333],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "hovertext": ["8.7x Slower", "2.1x Slower", "nanx Slower", "nanx Slower", "nanx Slower", "13.7x Slower", "nanx Slower", "nanx Slower", "nanx Slower", "nanx Slower"],
                    "marker": {"color": "rgba(255,193,30, 0.75)"},
                    "name": "Dask",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [42.37166666666667, 20.926666666666666, null, null, null, 36.968333333333334, null, null, null, null],
                    "type": "bar",
                    "textposition": "inside"
                }
            ],
            {
                "title": {"text": "TPCH 1000 Scale Factor - 4 Nodes (lower is better)"},
                "yaxis": {"title": {"text": "Time (minutes)"}},
                "xaxis": {"title": {"text": "TPCH Question"}},
                "uniformtext": {"minsize": 8, "mode": "hide"}
            },
            {"displayModeBar": false, "responsive": true}
        );
    }
</script>

* Only for queries that completed.

From the results we see that only Daft and Spark are able to complete all the questions. Dask completes less than a third and Modin is unable to complete any due to OOMs and cluster crashes. Since we can only compare to Spark here, we see that Daft is 3.5x faster including S3 IO. This shows that Daft and Spark are the only Dataframes in this comparison capable of processing data larger than memory, with Daft standing out as the significantly faster option.

1000 Scale Factor - Node Count Ablation

Finally, we compare how Daft performs on varying size clusters on the terabyte scale dataset. We run the same Daft TPC-H questions on the same dataset as the previous section but sweep the worker node count.

<script type="text/javascript">window.PlotlyConfig = {MathJaxConfig: 'local'};</script>
<script charset="utf-8" src="https://cdn.plot.ly/plotly-2.20.0.min.js"></script>

<script type="text/javascript">
    window.PLOTLYENV = window.PLOTLYENV || {};
    if (document.getElementById("8da53ffa-b330-43c6-b32b-a84051abed03")) {
        Plotly.newPlot(
            "8da53ffa-b330-43c6-b32b-a84051abed03",
            [
                {
                    "name": "1 Node",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [18.466666666666665, 34.7, 49.516666666666666, 37.583333333333336, 67.01666666666667, 12.133333333333333, 56.18333333333333, 68.68333333333334, 92.1, 57.63333333333333],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "name": "4 Node",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [4.85, 9.766666666666667, 12.933333333333334, 11.233333333333333, 17.616666666666667, 2.7, 15.15, 18.5, 22.833333333333332, 13.983333333333333],
                    "type": "bar",
                    "textposition": "inside"
                },
                {
                    "name": "8 Node",
                    "x": ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7", "Q8", "Q9", "Q10"],
                    "y": [2.6, 5.933333333333334, 6.583333333333333, 5.083333333333333, 10.2, 1.5, 7.95, 9.733333333333333, 16.666666666666668, 7.183333333333334],
                    "type": "bar",
                    "textposition": "inside"
                }
            ],
            {
                "title": {"text": "TPCH 1000 Scale Factor - Node Count vs Daft Query Time"},
                "yaxis": {"title": {"text": "Time (minutes)"}},
                "xaxis": {"title": {"text": "TPCH Question"}},
                "uniformtext": {"minsize": 8, "mode": "hide"}
            },
            {"displayModeBar": false, "responsive": true}
        );
    }
</script>

We note two interesting results here:

1. Daft can process 1TB+ of analytical data on a single 61GB instance without being distributed (16x more data than memory).
2. Daft query times scale linearly with the number of nodes (e.g. 4 nodes being 4 times faster than a single node). This allows for faster queries while maintaining the same compute cost!

Detailed Benchmarking Setup Copy Section

Benchmarking Code

Our benchmarking scripts and code can be found in the distributed-query-benchmarks GitHub repository.

• TPC-H queries for Daft were written by us.
• TPC-H queries for SparkSQL was adapted from this repository.
• TPC-H queries for Dask and Modin were adapted from these repositories for questions Q1-7 and Q8-10.

Infrastructure Copy Section

Our infrastructure runs on an EKS Kubernetes cluster.

• Driver Instance: i3.2xlarge
• Worker Instance: i3.2xlarge
• Number of Workers: 1/4/8
• Networking: All instances colocated in the same Availability Zone in the AWS us-west-2 region

Data

Data for the benchmark was stored in AWS S3.
No node-level caching was performed, and data is read directly from AWS S3 on every attempt to simulate realistic workloads.

• Storage: AWS S3 Bucket
• Format: Parquet
• Region: us-west-2
• File Layout: Each table is split into 32 (for the 100SF benchmark) or 512 (for the 1000SF benchmark) separate Parquet files. Parquet files for a given table have their paths prefixed with that table’s name, and are laid out in a flat folder structure under that prefix. Frameworks are instructed to read Parquet files from that prefix.
• Data Generation: TPC-H data was generated using the utilities found in the open-sourced Daft repository. This data is also available on request if you wish to reproduce any results!

Cluster Setup

Dask and Ray

To help us run the Distributed Dataframe libraries, we used Kubernetes for deploying Dask and Ray clusters.
The configuration files for these setups can be found in our open source benchmarking repository.

Our benchmarks for Daft and Modin were run on a KubeRay cluster, and our benchmarks for Dask was run on a Dask-on-Kubernetes cluster. Both projects are owned and maintained officially by the creators of these libraries as one of the main methods of deploying.

Spark

For benchmarking Spark we used AWS EMR, the official managed Spark solution provided by AWS. For more details on our setup and approach, please consult our Spark benchmarks README.

Logs

Connectors Index (`docs/connectors/index.md`) Copy Section

Data Connectors

Daft reads from and writes to a wide range of data sources — object storage, open table formats, catalogs, databases, files, and more. Please see Daft Connectors API docs for API details.

Object Storage Copy Chapter

Daft natively supports reading and writing data to major cloud object storage providers:

Table Formats Copy Chapter

Daft reads and writes the major open table formats.

Apache Hudi Copy Section

See also Apache Hudi for detailed integration.

Apache Iceberg Copy Section

See also Apache Iceberg for detailed integration.

Delta Lake Copy Section

See also Delta Lake for detailed integration.

Apache Paimon Copy Section

See also Apache Paimon for detailed integration.

Lance Copy Section

See also Lance for detailed integration.

Catalogs Copy Chapter

!!! warning "Warning"

These APIs are early in their development. Please feel free to [open feature requests and file issues](https://github.com/Eventual-Inc/Daft/issues/new/choose). We'd love to hear what you would like, thank you! 🤘

Catalogs are a centralized place to organize and govern your data. A catalog is often responsible for creating objects such as tables and namespaces, managing transactions, and access control. Most importantly, the catalog abstracts away physical storage details, letting you focus on the logical structure of your data without worrying about file formats, partitioning schemes, or storage locations.

Daft integrates with these catalog implementations:

Daft integrates with these via its Catalog and Table interfaces. These are high-level APIs to manage catalog objects (tables and namespaces), while also making it easy to leverage Daft's existing daft.read_ and df.write_ APIs for open table formats like Apache Iceberg and Delta Lake.

Example Copy Section

!!! note "Note"

These examples use the Iceberg Catalog from the [Daft Sessions](../configuration/sessions-usage.md) tutorial.

import daft

from daft import Catalog

# iceberg_catalog from the  'Sessions' tutorial
iceberg_catalog = load_catalog(...)

# create a daft catalog from the pyiceberg catalog instance
catalog = Catalog.from_iceberg(iceberg_catalog)

# verify
catalog
"""
Catalog('default')
"""

# we can read as a dataframe
catalog.read_table("example.tbl").schema()
"""
╭─────────────┬─────────╮
│ column_name ┆ type    │
╞═════════════╪═════════╡
│ x           ┆ Boolean │
├╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┤
│ y           ┆ Int64   │
├╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┤
│ z           ┆ Utf8    │
╰─────────────┴─────────╯
"""

# give a dataframe...
df = daft.from_pylist([{ "x": False, "y": -1, "z": "xyz" }])

# we can write to tables
catalog.write_table("example.tbl", df, mode="append")

# we can get also get table instances
t = catalog.get_table("example.tbl")

# see 'Working with Tables' for what we can do!
t
"""
Table('tbl')
"""

Usage Copy Section

This section covers detailed usage of the current APIs with some code snippets.

Working with Catalogs

The Catalog interface allows you to perform catalog actions like get_table and list_tables.

Example

import daft

from daft import Catalog, Table

# create a catalog from a pyiceberg catalog object
_ = Catalog.from_iceberg(pyiceberg_catalog)

# create a catalog from a unity catalog object
_ = Catalog.from_unity(unity_catalog)

# we can register various types as tables, note that all are equivalent
example_dict = { "x": [ 1, 2, 3 ] }
example_df = daft.from_pydict(example_dict)
example_table = Table.from_df("temp", example_df)

# create a catalog from a pydict mapping names to tables
catalog = Catalog.from_pydict(
    {
        "R": example_dict,
        "S": example_df,
        "T": example_table,
    }
)

# list available tables
# Note: Pattern syntax is catalog-dependent. Native/memory and Postgres catalogs use SQL LIKE syntax (%, _).
# Other catalogs like Iceberg use prefix matching.
catalog.list_tables(pattern=None)
"""
['R', 'S', 'T']
"""

# get a table by name
table_t = catalog.get_table("T")

#
table_t.show()
"""
╭───────╮
│ x     │
│ ---   │
│ Int64 │
╞═══════╡
│ 1     │
├╌╌╌╌╌╌╌┤
│ 2     │
├╌╌╌╌╌╌╌┤
│ 3     │
╰───────╯
"""

Working with Tables

The Table interface is a bridge from catalogs to dataframes. We can read tables into dataframes, and we can write dataframes to tables. You can work with a table independently of a catalog by using one of the factory methods, but it might not appear to provide that much utility over the existing daft.read_ and daft.write_ APIs. You would be correct in assuming that this is what is happening under the hood! The Table interface provides indirection over the table format itself and serves as a single abstraction for reading and writing that our catalogs can work with.

Examples

from daft import Table
from pyiceberg.table import StaticTable

# suppose you have a pyiceberg table
pyiceberg_table = StaticTable("metadata.json")

# we can make it a daft table to use daft's table APIS
table = Table.from_iceberg(pyiceberg_table)

# we can read a dataframe like `daft.read_iceberg(pyiceberg_table)`
df = table.read()

# you can also create temporary tables from dataframes
daft.create_temp_table("my_temp_table", daft.from_pydict({ ... }))

# these will be resolved just like other tables
df = daft.read_table("my_temp_table")

!!! note "Note"

Today you can read from `pyiceberg` and `daft.unity` table objects.

Reference Copy Section

!!! note "Note"

For complete documentation, please see the [Catalog & Table API docs](../api/catalogs_tables.md).

• [Catalog][daft.catalog.Catalog] - Interface for creating and accessing both tables and namespaces
• [Identifier][daft.catalog.Identifier] - Paths to objects e.g. catalog.namespace.table
• [Table][daft.catalog.Table] - Interface for reading and writing dataframes

Databases Copy Chapter

Google Cloud Bigtable Copy Section

!!! warning "Experimental"

This connector is experimental and the API may change.

See also Bigtable for detailed integration.

ClickHouse Copy Section

See also ClickHouse for detailed integration.

PostgreSQL Copy Section

See also Postgres for detailed integration.

SQL Databases Copy Section

See also SQL Databases for detailed integration.

Turbopuffer Copy Section

See also Turbopuffer for detailed usage.

Files Copy Chapter

File References Copy Section

See also Files for detailed usage.

CSV Copy Section

JSON Copy Section

Parquet Copy Section

Text and Blob Copy Section

See also Text Files and Blob Files for detailed usage.

WARC Copy Section

ORC Copy Section

See also ORC Files for schema and file-reading behavior.

WebDataset Copy Section

See also WebDataset for format details and lazy media reads.

Video Copy Section

Generic File Source Options Copy Section

Options shared across the file-based readers above (compression, schema hints, and more). See Generic File Source Options for the full list.

Other Sources Copy Chapter

Apache Kafka Copy Section

!!! warning "Experimental"

This connector is experimental. Currently only bounded batch reads are supported — there is no streaming/unbounded mode and no offset commit management.

See also Apache Kafka for detailed integration.

MCAP Copy Section

!!! warning "Experimental"

This connector is experimental. See [MCAP](mcap.md) for details.

See also MCAP for detailed integration.

Hugging Face Datasets Copy Section

See also Hugging Face Datasets for detailed usage.

Custom Copy Chapter

Custom Connectors Copy Section

Implement your own source or sink to read from and write to systems Daft does not support natively:

See also Custom Connectors for detailed usage.

Custom Catalogs Copy Section

Implement your own catalog by extending Daft's Catalog and Table interfaces. See Custom Catalogs for detailed usage.

In-Memory Copy Chapter

Create DataFrames directly from in-memory Python objects:

Custom Code Index (`docs/custom-code/index.md`) Copy Section

Running Custom Python Code in Daft

When Daft's built-in functions aren't sufficient for your needs, the @daft.func and @daft.cls decorators let you run your own Python code over each row of data. For custom aggregations, @daft.udaf lets you define aggregate functions with a three-stage pipeline. Simply decorate a Python function or class, and it becomes usable in Daft DataFrame operations.

If you are looking to migrate from Legacy UDFs to Daft's next generation UDF API, please see our migration guide.

Quick Example Copy Chapter

import daft

@daft.func
def add_and_format(a: int, b: int) -> str:
    return f"Sum: {a + b}"

df = daft.from_pydict({"x": [1, 2, 3], "y": [4, 5, 6]})
df = df.select(add_and_format(df["x"], df["y"]))
df.show()

╭───────────╮
│ x         │
│ ---       │
│ Utf8      │
╞═══════════╡
│ Sum: 5    │
├╌╌╌╌╌╌╌╌╌╌╌┤
│ Sum: 7    │
├╌╌╌╌╌╌╌╌╌╌╌┤
│ Sum: 9    │
╰───────────╯

Function Variants Copy Chapter

Daft supports multiple function variants to optimize for different use cases:

• Row-wise (default): Regular Python functions process one row at a time
• Async row-wise: Async Python functions process rows concurrently
• Generator: Generator functions produce multiple output rows per input row
• Batch: Process entire batches of data with daft.Series for high performance

Daft automatically detects which variant to use for regular functions based on your function signature. For batch functions, you must use the @daft.func.batch decorator.

End to End Example with Return Type Inference Copy Chapter

/* Detailed source-code truncated for AI context efficiency. */

Distributed Index (`docs/distributed/index.md`) Copy Section

Distributed Execution

Daft is designed to scale seamlessly from your laptop to distributed clusters. By default, Daft runs using your local machine's resources with the native runner (codenamed Swordfish), but when you need more compute power, you can easily scale to distributed execution across multiple machines.

There are a few ways to run Daft on multiple machines:

• For the easiest way to get started, check out our guide for running on Kubernetes
• Alternatively, if you have an existing Ray setup, check out our guide for running on Ray

Examples Index (`docs/examples/index.md`) Copy Section

Examples

• Hand Tracking for Physical AI

  Read a LeRobot dataset, run hand tracking as a Daft UDF, and score the results against EgoDex ground truth.

• Reward Scoring for Physical AI

  Read LIBERO episodes with Daft's LeRobot reader and score each one with a reward model - per-frame task progress plus success probability - then filter low-quality episodes with a Daft query.

• Motion Trimming for Physical AI

  Read DROID episodes with Daft's LeRobot reader, score per-frame motion from the robot's own joint positions - no video decode - and reduce it to one trim window per episode.

• Multimodal Structured Outputs: Evaluating Image Understanding

  Leverage image ablation to analyze textual bias in image understanding datasets.

• Voice AI Analytics with Faster-Whisper and embed_text

  Transcribe audio files into segments with timestamps and embed content.

• MinHash Deduplication on Common Crawl

  Clean web text at scale with MinHash, LSH Banding, and Connected Components.

• Getting Started with Common Crawl

  Daft provides a simple, performant, and responsible way to access Common Crawl data.

• Audio Transcription with Whisper

  Effortlessly transcribe audio to text at scale.

• Build a 100% GPU Utilization Text Embedding Pipeline featuring spaCy and Turbopuffer

  Generate and store millions of text embeddings in vector databases using distributed GPU processing and state-of-the-art models.

• Generate Images with Stable Diffusion

  Open Source image generation model on your own GPUs using Daft UDFs.

• Daft's Four UDF Pattern Tutorial

  One notebook, four UDF patterns, one dataset. Row-wise, generator, async, and stateful -- learn when to use each.

• Window Functions: The Great Chocolate Race

  Explore how window functions can reduce complex joins and groupby's to just a few simple operations.

• Running LLMs on the Red Pajamas Dataset

  Perform similarity search on Stack Exchange questions using language models and embeddings.

!!! tip "More Examples"
For more examples, check out our new daft-examples repository!

Extensions Overview (`docs/extensions/overview.md`) Copy Section

Extensions

Daft extensions are reusable libraries that add domain-specific functionality
on top of Daft without requiring every specialized function, datatype,
integration, or workflow to live directly in Daft core.

There are two broad ways to build extensions:

• Python UDF-based extensions: Python packages built on Daft's
  custom-code APIs, including
  @daft.func, @daft.func.batch,
  @daft.cls, and @daft.method.batch.
• Native ABI extensions: shared-library-backed packages that use Daft's
  Arrow C Data Interface ABI for lower-level, vectorized native execution.

Both models can expose clean Python APIs that feel like ordinary Daft
expression functions. The implementation details can stay hidden behind
functions, classes, and expressions.

Python UDF-Based Extensions Copy Chapter

The fastest way to build a Daft extension is often pure Python. Using
@daft.func, @daft.func.batch,
@daft.cls, and @daft.method.batch,
contributors can package reusable Python logic that runs inside Daft's
distributed execution engine.

This is ideal for libraries that orchestrate existing Python ecosystems,
external services, ML models, GPUs, data maintenance tasks, or domain-specific
workflows. These extensions do not require a native shared library or the
Arrow C ABI. They are normal Python packages that expose higher-level Daft
APIs.

daft-lance is an example of
this model. It extends Daft with Lance-specific distributed operations such as
file compaction, scalar indexing, column merging, and REST catalog operations.
Internally, it uses Daft's Python UDF and class-UDF APIs to distribute Lance
tasks across a Daft query, while users interact with functions like
compact_files, create_scalar_index, and merge_columns_df.

This pattern also exists inside Daft itself. The
daft.functions.ai module exposes high-level functions like
prompt,
embed_text,
embed_image,
classify_text, and
classify_image. To users, these look
like normal Daft expression functions. Under the hood, they use Daft's UDF and
class-UDF machinery, including batching, concurrency controls, retries, and
GPU resource hints.

The file APIs follow a similar product pattern. daft.File
values and helpers like file,
audio_file,
video_file,
file_path, and
file_size give users expression-level
building blocks for working with files. Those file objects can then flow into
Python UDFs, model pipelines, and domain-specific libraries without users
needing to think about execution details.

Native ABI Extensions Copy Chapter

For contributors who need lower-level, vectorized performance, Daft also
supports native extensions through a C ABI based on the
Arrow C Data Interface.

Native ABI extensions can add high-performance scalar functions, aggregate
functions, Python expression wrappers, and extension-backed datatypes. They
ship as pip-installable Python packages that bundle a native shared library.
Users import the package, load it into a Daft Session,
and call ordinary Python wrappers in their DataFrame expressions.

Rust currently has the most ergonomic SDK through
daft-ext,
while C++ is demonstrated through the raw ABI. Other systems languages are
possible if they can produce a shared library, export the expected C ABI, and
read/write Arrow C Data Interface arrays.

Daft's own examples include:

• examples/hello:
  a minimal Rust native extension that registers a greet scalar function.
• examples/dvector:
  a pgvector-style native extension for vector distance functions such as
  l2_distance, cosine_distance, inner_product, and jaccard_distance.
• examples/hello_cpp:
  a pure C++ native extension using Apache Arrow C++ and the raw Daft C ABI.

What Can Contributors Build? Copy Chapter

Contributors can build:

• Pure-Python UDF extension libraries using @daft.func,
  @daft.func.batch, @daft.cls, and
  @daft.method.batch
• Distributed task libraries like
  daft-lance
• GPU-backed model inference extensions
• AI, file, media, and multimodal processing libraries that hide UDFs behind
  expression APIs
• Native scalar functions
• Native aggregate functions / UDAFs
• Python expression wrappers
• Extension-backed logical datatypes such as
  DataType.extension(...)

Where to Start Copy Chapter

• Browse installable projects in Community Extensions.
• Learn how to build a native extension in the Authoring Guide.
• Browse projects using Daft in Built on Daft.
• Learn more about Daft's Python UDFs.

Modalities Overview (`docs/modalities/overview.md`) Copy Section

Modalities Overview

Daft is designed to work with any modality.

Artificial Intelligence now natively understands text, images, audio, video, and documents, but legacy engines were never designed to feed these formats to large models. Daft closes that gap, giving you one distributed engine that processes any modality, respects memory limits, and keeps GPUs fed so you can build the pipeline once and scale it anywhere.

• 🔠 Text

  Normalize, chunk, dedupe, prompt, and embed text data.

• 🌄 Images

  Work with visual data and image processing. Use daft.ImageFile for lazy metadata extraction and filtered decoding.

• 🔉 Audio

  Read, extract metadata, resample audio files.

• 🎥 Video

  Working with video files and metadata.

• 📄 Documents

  Extract text and image data from PDF documents.

• {} JSON and Nested Data

  Parse, query, and manipulate semi-structured and hierarchical data.

• ⊹ Embeddings

  Generate vector representations for RAG and AI search.

• 📁 Generic Files and URLs

  Take advantage of Daft's built-in URL functions and daft.File types

**[Custom Modalities](custom.md)** Copy Section

The most important modality might be one we haven't explored yet. Daft makes it easy to define your own modality with custom connectors to read and write any kind of data, and use user-defined functions to process custom Python code efficiently and reliably at scale.

Sql Index (`docs/sql/index.md`) Copy Section

Daft SQL

SQL Reference

Welcome to Daft SQL Reference.

Daft's SQL dialect closely follows both DuckDB and PostgreSQL.

Example Copy Chapter

Please see Sessions and Catalogs for a detailed look at connecting data sources to Daft SQL.

import daft

from daft import Session

# create a session
sess = Session()

# create temp tables
sess.create_temp_table("T", daft.from_pydict({ "a": [ 0, 1 ] }))
sess.create_temp_table("S", daft.from_pydict({ "b": [ 1, 0 ] }))

# execute sql
sess.sql("SELECT * FROM T, S").show()
"""
╭───────┬───────╮
│ a     ┆ b     │
│ ---   ┆ ---   │
│ Int64 ┆ Int64 │
╞═══════╪═══════╡
│ 0     ┆ 1     │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
│ 1     ┆ 1     │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
│ 0     ┆ 0     │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
│ 1     ┆ 0     │
╰───────┴───────╯
"""

Usage Copy Chapter

SQL with DataFrames Copy Section

Daft's [daft.sql][daft.sql.sql.sql] function automatically detects any [daft.DataFrame][daft.DataFrame] objects in your current Python environment to let you query them easily by name.

=== "⚙️ SQL"
```python
# Note the variable name my_special_df
my_special_df = daft.from_pydict({"A": [1, 2, 3], "B": [1, 2, 3]})

# Use the SQL table name "my_special_df" to refer to the above DataFrame!
sql_df = daft.sql("SELECT A, B FROM my_special_df")

sql_df.show()
```

╭───────┬───────╮
│ A     ┆ B     │
│ ---   ┆ ---   │
│ Int64 ┆ Int64 │
╞═══════╪═══════╡
│ 1     ┆ 1     │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
│ 2     ┆ 2     │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌┤
│ 3     ┆ 3     │
╰───────┴───────╯

(Showing first 3 of 3 rows)

In the above example, we query the DataFrame called "my_special_df" by simply referring to it in the SQL command. This produces a new DataFrame sql_df which can natively integrate with the rest of your Daft query. You can also use table functions to query sources directly.

=== "🐍 Python"
python daft.sql("SELECT * FROM read_parquet('s3://...')") daft.sql("SELECT * FROM read_iceberg('s3://.../metadata.json')")

Table Function Options Copy Section

Table functions take the path as the first positional argument, plus any of the options
below. Named arguments use either => or :=:

=== "🐍 Python"
python daft.sql("SELECT * FROM read_csv('data.csv', has_headers => false)") daft.sql("SELECT * FROM read_csv('data.csv', has_headers := false)")

read_parquet, read_csv, read_json, and read_deltalake also accept the path itself
as a named path argument:

=== "🐍 Python"
python daft.sql("SELECT * FROM read_csv(path => 'data.csv')")

A list of paths is written as a SQL array:

=== "🐍 Python"
python daft.sql("SELECT * FROM read_parquet(['a.parquet', 'b.parquet'])")

These mirror the corresponding Python reader arguments — see
[read_parquet][daft.read_parquet], [read_csv][daft.read_csv],
[read_json][daft.read_json], [read_iceberg][daft.read_iceberg], and
[read_deltalake][daft.read_deltalake] for what each one does, and
Generic File Source Options
for ignore_corrupt_files in depth.

A schema is given as a struct literal mapping column names to type names:

=== "🐍 Python"
python daft.sql(""" SELECT * FROM read_csv('data.csv', schema := {'a': 'int64', 'b': 'string'}) """)

For Iceberg, snapshot_id, branch, and tag select which version to read and are
mutually exclusive:

=== "🐍 Python"
python daft.sql("SELECT * FROM read_iceberg('/warehouse/db/t/metadata/v3.metadata.json', branch => 'audit')")

ignore_corrupt_files skips unreadable files instead of failing the query, and
df.skipped_corrupt_files reports what was skipped once the DataFrame is materialized:

=== "🐍 Python"
python df = daft.sql("SELECT * FROM read_csv('s3://my-bucket/data/**/*.csv', ignore_corrupt_files => true)") df.collect() print(df.skipped_corrupt_files)

SQL Expressions Copy Section

SQL has the concept of expressions as well. Here is an example of a simple addition expression, adding columns A and B in SQL to produce a new column C.

We also present here the equivalent query for SQL and DataFrame. Notice how similar the concepts are!

=== "⚙️ SQL"
python df = daft.from_pydict({"A": [1, 2, 3], "B": [1, 2, 3]}) df = daft.sql("SELECT A + B as C FROM df") df.show()

=== "🐍 Python"
``` python
expr = (daft.col("A") + daft.col("B")).alias("C")

df = daft.from_pydict({"A": [1, 2, 3], "B": [1, 2, 3]})
df = df.select(expr)
df.show()
```

╭───────╮
│ C     │
│ ---   │
│ Int64 │
╞═══════╡
│ 2     │
├╌╌╌╌╌╌╌┤
│ 4     │
├╌╌╌╌╌╌╌┤
│ 6     │
╰───────╯

(Showing first 3 of 3 rows)

In the above query, both the SQL version of the query and the DataFrame version of the query produce the same result.

Under the hood, they run the same Expression col("A") + col("B")!

One really cool trick you can do is to use the [daft.sql_expr][daft.sql.sql.sql_expr] function as a helper to easily create Expressions. The following are equivalent:

=== "⚙️ SQL"
python sql_expr = daft.sql_expr("A + B as C") print("SQL expression:", sql_expr)

=== "🐍 Python"
python py_expr = (daft.col("A") + daft.col("B")).alias("C") print("Python expression:", py_expr)

SQL expression: col(A) + col(B) as C
Python expression: col(A) + col(B) as C

This means that you can pretty much use SQL anywhere you use Python expressions, making Daft extremely versatile at mixing workflows which leverage both SQL and Python.

As an example, consider the filter query below and compare the two equivalent Python and SQL queries:

=== "⚙️ SQL"
```python
df = daft.from_pydict({"A": [1, 2, 3], "B": [1, 2, 3]})

# Daft automatically converts this string using `daft.sql_expr`
df = df.where("A < 2")

df.show()
```

=== "🐍 Python"
``` python
df = daft.from_pydict({"A": [1, 2, 3], "B": [1, 2, 3]})

# Using Daft's Python Expression API
df = df.where(df["A"] < 2)

df.show()
```

╭───────┬───────╮
│ A     ┆ B     │
│ ---   ┆ ---   │
│ Int64 ┆ Int64 │
╞═══════╪═══════╡
│ 1     ┆ 1     │
╰───────┴───────╯

(Showing first 1 of 1 rows)

Pretty sweet! Of course, this support for running Expressions on your columns extends well beyond arithmetic as we'll see in the next section on SQL Functions.

SQL Functions Copy Section

SQL also has access to all of Daft's powerful [daft.Expression][daft.Expression] functionality through SQL functions.

However, unlike the Python Expression API which encourages method-chaining (e.g. col("a").download().decode_image()), in SQL you have to do function nesting instead (e.g. "image_decode(url_download(a))").

Here is an example of an equivalent function call in SQL vs Python:

=== "⚙️ SQL"
python df = daft.from_pydict({"urls": [ "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", ]}) df = daft.sql("SELECT image_decode(url_download(urls)) FROM df") df.show()

=== "🐍 Python"
python df = daft.from_pydict({"urls": [ "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", "https://user-images.githubusercontent.com/17691182/190476440-28f29e87-8e3b-41c4-9c28-e112e595f558.png", ]}) df = df.select(daft.col("urls").download().decode_image()) df.show()

╭──────────────╮
│ urls         │
│ ---          │
│ Image[MIXED] │
╞══════════════╡
│ <Image>      │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ <Image>      │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ <Image>      │
╰──────────────╯

(Showing first 3 of 3 rows)

SUMMARY (`docs/SUMMARY.md`) Copy Section

• Guide
  • Overview
  • Quickstart
  • Installation
  • AI Functions
    • Overview
    • Prompt
    • Embed
    • Classify
    • Jev
    • Providers
  • Modalities
    • Overview
    • Text
    • Images
    • Audio
    • Videos
    • Documents
    • JSON and Nested Data
    • Files and URLs
    • Tensors
    • Embeddings
    • Custom Modalities
  • User Defined Functions
    • Overview
    • Functions
    • Classes & Methods
    • Aggregate UDFs
    • Working with GPUs
    • Legacy UDF Migration Guide
    • Legacy UDFs
  • Common Use Cases
    • Batch Inference
  • Datasets
    • Common Crawl
    • LeRobot
    • DROID
  • Data Connectors
    • Overview
    • Object Storage
      • AWS S3
      • Azure Blob Store
      • Google Cloud Storage
      • COS (Tencent Cloud)
      • GooseFS
    • Table Formats
      • Apache Hudi
      • Apache Iceberg
      • Delta Lake
      • Apache Paimon
      • Lance
    • Catalogs
      • AWS Glue
      • AWS S3 Tables
      • Apache Gravitino
      • Unity Catalog (Databricks)
    • Databases
      • Bigtable
      • ClickHouse
      • Postgres
      • SQL Databases
      • Turbopuffer
    • Files
      • Files
      • Text Files
      • Blob Files
      • WebDataset
      • ORC Files
      • Generic File Source Options
    • Other Sources
      • Apache Kafka
      • MCAP
      • Hugging Face Datasets
    • Custom
      • Custom Connectors
      • Custom Catalogs
  • Extensions
    • Overview
    • Community Extensions
    • Built on Daft
    • Authoring Guide
  • Architecture
  • Scaling & Performance
    • Scaling Out and Deployment
      • Distributed Execution
      • Running on Kubernetes
      • Running on Ray
    • Optimization
      • Managing Memory Usage
      • Partitioning and Batching
      • Shuffle Algorithms
      • Join Strategies
  • Checkpointing
  • Observability
    • Dashboard
    • Progress Indicators
    • Logging
    • Telemetry
  • Sessions, Catalogs, and Tables
  • Resources
    • Roadmap
    • Benchmarks
    • Community <sup>↗</sup>
    • Release Notes <sup>↗</sup>
    • Usage Telemetry
• Examples
  • Examples
  • Hand Tracking for Physical AI
  • Reward Scoring for Physical AI
  • Motion Trimming for Physical AI
  • Multimodal Structured Outputs: Evaluating Image Understanding
  • Voice AI Analytics with Faster-Whisper and embed_text
  • Web Text Deduplication
  • Audio Transcription
  • Generate Text Embeddings for Turbopuffer
  • Running LLMs on the Red Pajamas Dataset
  • Generate Images from Text with Stable Diffusion
  • Querying Image Data
  • MNIST Digit Classification
  • UDF Patterns
  • Window Functions
  • Working with Common Crawl Data
  • Document Processing
• Python API
  • Python API
  • AI
  • I/O
  • DataFrame
  • Datasets
  • Expressions
  • Functions
  • User-Defined Functions
  • Data Types
    • DataType
    • File Types
    • Type Conversions
    • Casting
  • Window
  • Sessions
  • Catalogs & Tables
  • Schema
  • Aggregations
  • Series
  • Configuration
  • Miscellaneous
• SQL Reference
  • SQL Reference
  • Statements
    • SELECT
    • DESCRIBE
    • SHOW
    • USE
  • Data Types
  • Identifiers
  • Window Functions
• Contributing
  • Overview
  • Development Guide
  • Contributing AI Functions
• Daft Skills

Install (`docs/install.md`) Copy Section

How to Install Daft

This guide helps you install Daft based on your specific needs and environment.

For most users, install Daft with a single command:

pip install -U daft

Optional Dependencies Copy Chapter

Depending on your use case, you may need to install Daft with additional dependencies.

<h4>Use Cases:</h4>


  <label class="checkbox-item">
    
    
    
      **Hugging Face** `huggingface`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **OpenAI** `openai`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Transformers** `transformers`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Distributed Computing on Ray** `ray`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Turbopuffer** `turbopuffer`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **AWS Glue or AWS S3 Tables** `aws`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Apache Iceberg** `iceberg`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Delta Lake** `deltalake`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Unity Catalog (Databricks)** `unity`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **LanceDB** `lance`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **ClickHouse** `clickhouse`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Azure Blob Storage** `azure`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Google Gemini** `google`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Apache Kafka** `kafka`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Apache Gravitino** `gravitino`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **PostgreSQL** `postgres`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **SQL Databases** `sql`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **pandas** `pandas`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **NumPy** `numpy`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Audio** `audio`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **Video** `video`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **HDF5** `hdf5`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **HTTP** `http`
    
  </label>

  <label class="checkbox-item">
    
    
    
      **MCAP** `mcap`
    
  </label>





<h4>Installation Command:</h4>

You can also install Daft with all extra dependencies:

pip install -U "daft[all]"

Nightly Builds Copy Chapter

Nightly builds are published daily from the latest main branch. These are useful for testing upcoming features and bug fixes before they are released on PyPI.

pip install daft --pre --extra-index-url https://nightly.daft.ai

!!! warning "Stability"
Nightly builds may contain unstable or experimental changes. They are not recommended for production use.

Troubleshooting Legacy CPU Support Copy Chapter

If you encounter Illegal instruction errors, your CPU may lack support for advanced instruction sets like AVX. Use the LTS version instead:

pip install -U daft-lts

!!! warning "Performance Impact"
The LTS version uses limited CPU instructions and cannot leverage vectorized operations, resulting in slower performance. Only use this if the standard package fails to run.

Roadmap (`docs/roadmap.md`) Copy Section

Roadmap

Last updated: March 2026

This roadmap outlines the big picture of what the Daft team plans to work on in the coming year, as well as some of the features to expect from these.

Please note that items on this roadmap are subject to change at any time. If there are features you would like to implement, we welcome and encourage open-source contributions. Our team is happy to provide guidance, help scope the work, and review PRs.

Feel free to open an issue or PR on GitHub ↗ or join our Daft Slack Community ↗.

Performance Copy Chapter

1. Performant large scale shuffles - Support for native shuffling services for data transfer in our distributed engine (Flotilla). This includes a specialized peer-to-peer service based on Arrow Flight RPC. (discussion #6472 ↗)
2. Distributed engine optimization - Optimizing our distributed engine (Flotilla) to dynamically schedule and execute tasks to minimize worker setup overhead and contention, add dynamic partitioning.
3. Memory management - Tracking and managing memory usage within our local runner to reduce chances of OOMs, inform backpressure and dynamic batching.
4. Hash-based operator improvements - Improve partitioning and hash-table building techniques within various hash-based operators, particularly groupby aggregations and hash joins in low-cardinality cases.

Key Features Copy Chapter

1. Checkpointing - Ability to checkpoint and resume long running workloads from a given checkpoint. (discussion #6446)
2. Kubernetes support - Distributed Daft running directly on Kubernetes.
3. Multiple distributed backends - Enable Daft to run distributed on multiple backends with a generic interface and continued support of Ray as a backend.

Observability Copy Chapter

1. Daft Dashboard - Observability dashboard with support for Flotilla, providing cluster, task and partition level observability.
2. Improved Debugging - Support for capturing and retrieving debugging information and run history.
3. Memory and CPU observability - Ability to capture and retrieve resource utilization metrics.

Extensibility Copy Chapter

1. Native extensions - Support for Daft native extensions via the Arrow C interfaces; enabling developers to extend Daft in any language which supports C-based interop.
2. Data Source refactor - Simplify the source pipeline to make adding native Rust sources straightforward.
3. Arrow2 deprecation (discussion #5741)

Future Work Copy Chapter

The following features are in consideration, but are not currently on our roadmap. We're sharing these to highlight opportunities for open source contributions, invite discussion around implementation approaches, and provide further visibility. These features have been tagged with help wanted and good first issue on Daft repo ↗.

1. Improved Delta Lake support (see roadmap for Delta Lake ↗)
   • Support for reading tables with deletion vectors (issue #1954 ↗)
   • Support for reading tables with column mappings (issue #1955 ↗)
2. VARIANT type - Support for the VARIANT type into Daft's type system which is compatible with the parquet VARIANT.
3. Result<T> type - Add a Result<T> type (Either) to Daft's type system to support graceful handling of fallible operations.

If you are interested in working on any of these features, feel free to open an issue or start a discussion on GitHub ↗ or join our Daft Slack Community ↗. Our team can provide technical direction and help scope the work appropriately. Thank you in advance 💜

Robots (`docs/robots.txt`) Copy Section

User-agent: *

Disallow: /en/
Allow: /en/stable

Sitemap: https://docs.daft.ai/en/stable/sitemap.xml

Skills (`docs/skills.md`) Copy Section

Daft Skills

Daft ships with built-in skills that help AI coding agents work effectively with Daft.

Available skills Copy Chapter

[daft-udf-tuning](https://github.com/Eventual-Inc/Daft/blob/main/.claude/skills/daft-udf-tuning/SKILL.md) Copy Section

Optimize User-Defined Functions for performance.

Triggers: GPU inference setup, slow UDFs, async or batch processing questions.

What it helps with:

• Choosing between @daft.func (stateless), @daft.cls (stateful), and @daft.func.batch (vectorized)
• Async I/O patterns for web APIs
• GPU batch inference with PyTorch/model loading via @daft.cls(gpus=N)
• Tuning max_concurrency, batch_size, gpus, and into_batches(N)

[daft-distributed-scaling](https://github.com/Eventual-Inc/Daft/blob/main/.claude/skills/daft-distributed-scaling/SKILL.md) Copy Section

Scale single-node workflows to distributed execution.

Triggers: Performance optimization, handling large datasets, distributed execution.

What it helps with:

• Choosing between shuffle (repartition(N)) and streaming (into_batches(N)) strategies
• Calculating optimal partition counts for your cluster
• Avoiding OOM errors with heavy data (images, tensors)
• Advanced tuning formulas for repartitioning and streaming

[daft-docs-navigation](https://github.com/Eventual-Inc/Daft/blob/main/.claude/skills/daft-docs-navigation/SKILL.md) Copy Section

Navigate Daft's documentation for APIs, concepts, and examples.

Triggers: General questions about Daft APIs, concepts, or searching the docs.

What it helps with:

• Navigating key doc locations: API reference, optimization, distributed, UDFs, connectors
• Searching for API usage in the docs/ directory
• Browsing the docs structure via docs/SUMMARY.md

Installation Copy Chapter

Already cloned Daft (automatic) Copy Section

If you're working inside a Daft clone, the skills are already available. No setup needed - Claude Code picks them up from .claude/skills/.

Install with `npx skills add` Copy Section

To use these Daft skills outside the repo (across many agents including Claude Code, Cursor, Codex, Gemini CLI, Copilot, and more):

# Install all skills
npx skills add Eventual-Inc/Daft

# Install a specific skill
npx skills add Eventual-Inc/Daft --skill daft-udf-tuning

Follow the prompts to choose skills, agents, scope (project or global), and whether to symlink or copy.

Claude Code plugin Copy Section

These Daft skills are also available as a Claude Code plugin.

claude plugin marketplace add Eventual-Inc/Daft
claude plugin install daft@Eventual-Inc

Telemetry (`docs/telemetry.md`) Copy Section

Telemetry

To help core developers improve Daft, we collect non-identifiable statistics on Daft usage in order to better understand how Daft is used, common bugs and performance bottlenecks. Data is collected via Scarf and osstelemetry.io.

We take the privacy of our users extremely seriously, and telemetry in Daft is built to be:

1. Easy to opt-out: To disable telemetry, set the environment variable DO_NOT_TRACK=true

2. Non-identifiable: No session IDs or user identifiers are collected

3. Metadata-only: We do not collect any of our users' proprietary code or data

We do not sell or buy any of the data that is collected in telemetry.

!!! info "Daft telemetry is enabled in versions >= v0.0.21"

What data do we collect? Copy Chapter

To audit what data is collected, please see scarf_telemetry.py.

In short, we collect the following:

1. On import, we track the version of Daft, OS, Python version, and system architecture.
2. On DataFrame executions, we track the same system information as we do on imports, in addition to the runner being used, i.e. native or Ray.

Ai Functions Classify (`docs/ai-functions/classify.md`) Copy Section

Text and Image Classification with Daft

Daft provides powerful classification functions for both text and images, making it easy to label and categorize your data at scale. Whether you're working with sentiment analysis, content categorization, or visual recognition tasks, Daft's classification functions integrate seamlessly with popular models from Hugging Face and other providers.

!!! tip "Choosing the Right Approach"
- Use classify_text and classify_image for fast, efficient classification with pre-defined labels
- Use prompt for more flexible classification that requires explanations or complex reasoning
- For custom models, Daft supports both Hugging Face Transformers and OpenAI-compatible providers

Text Classification Copy Chapter

The classify_text function provides fast, efficient text classification using zero-shot models or fine-tuned classifiers. It's ideal for tasks like sentiment analysis, topic categorization, and intent detection where you have predefined labels.

import daft
from daft.functions import classify_text

df = daft.from_pydict({
    "text": [
        "One day I will see the world",
        "I've always enjoyed preparing dinner for my family",
    ],
})

df = df.with_column(
    "label",
    classify_text(
        daft.col("text"),
        labels=['travel', 'cooking', 'dancing'],
        provider="transformers",
        model="facebook/bart-large-mnli",
    ),
)

df.show()

╭────────────────────────────────┬─────────╮
│ text                           ┆ label   │
│ ---                            ┆ ---     │
│ String                         ┆ String  │
╞════════════════════════════════╪═════════╡
│ One day I will see the world   ┆ travel  │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┤
│ I've always enjoyed preparing… ┆ cooking │
╰────────────────────────────────┴─────────╯

(Showing first 2 of 2 rows)

Image Classification Copy Chapter

The classify_image function uses vision models like CLIP for zero-shot image classification. This example shows a complete pipeline including image preprocessing steps:

import daft
from daft.functions import classify_image, decode_image, convert_image, resize

df = (
    # Discover a few images from HuggingFace
    daft.from_glob_path("hf://datasets/datasets-examples/doc-image-3/images")
    # Read the 4 PNG, JPEG, TIFF, WEBP Images
    .with_column("image_bytes", daft.col("path").download())
    # Decode the image bytes into a daft Image DataType
    .with_column("image_type", decode_image(daft.col("image_bytes")))
    # Convert Image to RGB and resize the image to 224x224
    .with_column(
        "image_resized", resize(convert_image(daft.col("image_type"), "RGB"), 224, 224)
    )
    # Classify the image
    .with_column(
        "image_label",
        classify_image(
            daft.col("image_resized"),
            provider="transformers",
            labels=["bulbasaur", "catapie", "voltorb", "electrode"],
            model="openai/clip-vit-base-patch32",
        ),
    )
)

df.show()

/* Detailed source-code truncated for AI context efficiency. */

!!! note "When to Use Prompts for Classification"
Use classify_image and classify_text when you have a fixed set of labels and need fast, efficient classification. Use prompt when you need:

- Explanations for classification decisions
- Complex reasoning or multi-step logic
- Dynamic or unstructured label sets
- Additional context beyond a simple label

Flexible Classification with Prompts Copy Chapter

When you need more flexibility—such as generating explanations, handling complex reasoning, or working with unstructured labels—use the prompt function. This approach trades some speed for significantly more capability.

import os
import daft
from daft.functions import prompt

daft.set_provider(
    "openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ.get("OPENROUTER_API_KEY")
)

df = daft.from_pydict({
    "quote": [
        "I am going to be the king of the pirates!",
        "I'm going to be the next Hokage!",
    ],
})

df = (
    df
    .with_column(
        "response",
        prompt(
            daft.col("quote"),
            system_message="Classify the anime from the quote and return the show, character name, and explanation.",
            provider="openai",
            model="nvidia/nemotron-nano-9b-v2:free"
        )
    )
)

df.show(format="fancy", max_width=120)

╭───────────────────────────────────────────┬──────────────────────────────────────────────────────────────╮
│ quote                                     ┆ response                                                     │
╞═══════════════════════════════════════════╪══════════════════════════════════════════════════════════════╡
│ I am going to be the king of the pirates! ┆ **Show:** *One Piece*                                        │
│                                           ┆ **Character Name:** Monkey D. Luffy                          │
│                                           ┆ **Explanation:**                                             │
│                                           ┆ The quote "I am going to be the king…                        │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ I'm going to be the next Hokage!          ┆ **Show:** *Naruto*                                           │
│                                           ┆ **Character Name:** Naruto Uzumaki                           │
│                                           ┆ **Explanation:** The quote "I'm going to be the next Hokage… │
╰───────────────────────────────────────────┴──────────────────────────────────────────────────────────────╯
(Showing first 2 of 2 rows)

Ai Functions Embed (`docs/ai-functions/embed.md`) Copy Section

Generate Embeddings from Text and Images with Daft

Embeddings are dense vector representations of data that capture semantic meaning. Daft provides fast, scalable embedding generation for both text and images, with support for various providers including OpenAI, Transformers, and local models.

Text Embeddings Copy Chapter

The embed_text function generates embeddings from text using various models. You can use AI functions without explicitly configuring a provider—if you have an OPENAI_API_KEY environment variable set, simply specify provider="openai" and it will automatically create a provider in your session.

import daft
from daft.functions import embed_text

df = daft.from_pydict({"text": ["Hello World"]})

df = df.with_column(
    "embeddings",
    embed_text(
        daft.col("text"),
        provider="openai",
        model="text-embedding-3-small"
    )
)

df.show()

╭─────────────┬──────────────────────────╮
│ text        ┆ embeddings               │
│ ---         ┆ ---                      │
│ String      ┆ Embedding[Float32; 1536] │
╞═════════════╪══════════════════════════╡
│ Hello World ┆ ▆█▆▆▆▃▆▆▂▄▃▂▃▃▄▁▃▅▂▃▂▂▂▂ │
╰─────────────┴──────────────────────────╯

Local Embeddings with LM Studio Copy Section

For privacy-sensitive workloads or offline environments, you can generate embeddings locally with LM Studio. This lets you run embedding models like Nomic, Qwen, or Mistral on your own machine, taking advantage of hardware accelerators like Apple's Metal Performance Shaders (MPS).

import daft
from daft.functions import embed_text

df = daft.from_pydict({
    "text": ["Hello, world!"]
})

df = df.with_column(
    "embedding",
    embed_text(daft.col("text"), model="text-embedding-nomic-embed-text-v1.5", provider="lm_studio")
)

df.show()

╭───────────────┬──────────────────────────╮
│ text          ┆ embedding                │
│ ---           ┆ ---                      │
│ String        ┆ Embedding[Float32; 768]  │
╞═══════════════╪══════════════════════════╡
│ Hello, world! ┆ █▄▄▃▃▄▃▄▃▆▂▁▂▅▅▂▂▃▅▄▄▃▂▄ │
╰───────────────┴──────────────────────────╯

(Showing first 1 of 1 rows)

Provider-specific request parameters Copy Section

OpenAI-compatible embedding servers can accept additional JSON request properties through extra_body. For example,
vLLM can truncate long inputs from the right, keeping the first 4,096 tokens:

df = df.with_column(
    "embedding",
    embed_text(
        daft.col("text"),
        provider=provider,
        model="BAAI/bge-m3",
        extra_body={
            "truncate_prompt_tokens": 4096,
            "truncation_side": "right",
        },
    ),
)

Image Embeddings Copy Chapter

The embed_image function generates embeddings from images using vision models. Daft natively supports images with the Image DataType, making multimodal preprocessing straightforward with built-in utilities like decode_image, convert_image, and resize.

This example shows a complete pipeline: reading multiple image formats (PNG, JPEG, TIFF, WEBP) from HuggingFace, preprocessing them to the required format, and generating embeddings using Apple's aimv2-large-patch14-224-lit model.

import daft
from daft.functions import embed_image, decode_image, convert_image, resize

df = (
    # Discover a few images from HuggingFace
    daft.from_glob_path("hf://datasets/datasets-examples/doc-image-3/images")
    # Read the 4 PNG, JPEG, TIFF, WEBP Images
    .with_column("image_bytes", daft.col("path").download())
    # Decode the image bytes into a daft Image DataType
    .with_column("image_type", decode_image(daft.col("image_bytes")))
    # Convert Image to RGB and resize the image to 288x288
    .with_column("image_resized", resize(convert_image(daft.col("image_type"), "RGB"), 288, 288))
    # Embed the image
    .with_column(
        "image_embeddings",
        embed_image(
            daft.col("image_resized"),
            provider="transformers",
            model="apple/aimv2-large-patch14-224-lit"
        )
    )
)

# Show the dataframe
df.show()

/* Detailed source-code truncated for AI context efficiency. */

!!! tip "Image Preprocessing"

When working with images for embedding, it's important to preprocess them to the right size and format. Be sure to check the model card for specific requirements regarding image format, size, and normalization.

Semantic Search with Cosine Distance Copy Chapter

Once you have embeddings, you can use cosine distance to find similar items. This example demonstrates a simple RAG (Retrieval-Augmented Generation) pattern where we find the most relevant documents for a query.

import daft
from daft.functions import embed_text, cosine_distance

# Create a knowledge base with documents
documents = daft.from_pydict({
    "doc_id": [1, 2, 3, 4],
    "text": [
        "Python is a high-level programming language",
        "Machine learning models require training data",
        "Daft is a distributed dataframe library",
        "Embeddings capture semantic meaning of text",
    ]
})

# Embed all documents
documents = documents.with_column(
    "embedding",
    embed_text(daft.col("text"), provider="openai", model="text-embedding-3-small")
)

# Create a query
query = daft.from_pydict({
    "query_text": ["What is Daft?"]
})

# Embed the query
query = query.with_column(
    "query_embedding",
    embed_text(
        daft.col("query_text"),
        provider="openai",             # Requires OPENAI_API_KEY
        model="text-embedding-3-small"
    )
)

# Cross join to compare query against all documents
results = query.join(documents, how="cross")

# Calculate cosine distance (lower is more similar)
results = results.with_column(
    "distance",
    cosine_distance(daft.col("query_embedding"), daft.col("embedding"))
)

# Sort by distance and show top results
results = results.sort("distance").select("query_text", "text", "distance")
results.show()

╭───────────────┬────────────────────────────────┬─────────────────────╮
│ query_text    ┆ text                           ┆ distance            │
│ ---           ┆ ---                            ┆ ---                 │
│ String        ┆ String                         ┆ Float64             │
╞═══════════════╪════════════════════════════════╪═════════════════════╡
│ What is Daft? ┆ Daft is a distributed datafra… ┆ 0.36203420173101486 │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ What is Daft? ┆ Python is a high-level progra… ┆ 0.91644507004277    │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ What is Daft? ┆ Embeddings capture semantic m… ┆ 0.937416795759487   │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ What is Daft? ┆ Machine learning models requi… ┆ 0.9697065433855928  │
╰───────────────┴────────────────────────────────┴─────────────────────╯

(Showing first 4 of 4 rows)

!!! tip "Scaling Similarity Search"
For large-scale similarity search, consider:

- Pre-computing and storing embeddings to avoid re-generation
- Using approximate nearest neighbor (ANN) algorithms for huge datasets
- Filtering documents before computing similarity to reduce comparisons
- Writing results to a vector database like [Turbopuffer](../connectors/turbopuffer.md) for efficient retrieval

Ai Functions Jev (`docs/ai-functions/jev.md`) Copy Section

Filtering, Classifying, and Scoring Rows with Jev

Daft's jev functions let you ask plain-English questions about each row of a DataFrame: does this ticket come from an angry customer, which team should handle it, how luxurious is this product? Questions are answered by TypeSafe's Jev model, and the functions are modelled after DuckDB's jev community extension.

!!! tip "Choosing the Right Approach"
- Use jev and jev_prob to filter or rank rows by a condition that is easy to state but hard to code
- Use jev_choice and jev_score to label rows without hosting a model; use classify_text to run a classifier locally
- Use prompt when you need free-form text, explanations, or structured outputs

Setup Copy Chapter

Install Daft with the jev extra and set your TypeSafe API key:

pip install "daft[jev]"
export TYPESAFE_API_KEY="..."

Unlike the other AI functions, the jev functions don't use a provider. Each function takes its settings as keyword arguments, falling back to environment variables:

Settings are read when you build the expression, so workers on a distributed cluster don't need the environment variables.

Filtering with `jev` Copy Chapter

The jev function returns True for rows that satisfy a plain-English condition, which makes it a natural fit for where. To judge several columns at once, pack them into a struct with to_struct; the field names are visible to the model.

import daft
from daft.functions import jev, to_struct

df = daft.from_pydict({
    "first": ["Hans", "Mei", "Giulia", "Kwame"],
    "last": ["Müller", "Chen", "Rossi", "Mensah"],
})

df = df.where(
    jev(to_struct(daft.col("first"), daft.col("last")), "the name is European")
)

df.show()

A row counts as True when its probability is at least threshold, which defaults to 0.5. Raise it to keep only confident matches:

df.where(jev(daft.col("ticket"), "the customer is angry", threshold=0.9))

Probabilities with `jev_prob` Copy Chapter

The jev_prob function returns the probability behind jev, so you can rank rows or pick a threshold after looking at the data. jev(row, condition, threshold) is equivalent to jev_prob(row, condition) >= threshold.

import daft
from daft.functions import jev_prob

df = daft.from_pydict({
    "ticket": [
        "This is the third time you've double charged me. I am furious and cancelling today!",
        "Thanks so much, the new dashboard is lovely.",
        "I was billed twice for my subscription this month.",
        "The app crashes every time I try to log in.",
    ],
})

df = df.with_column("p_angry", jev_prob(daft.col("ticket"), "the customer is angry"))

df.sort("p_angry", desc=True).show()

Classification with `jev_choice` Copy Chapter

The jev_choice function picks the most probable of a list of options as the answer to a question about each row. It takes between 1 and 255 options, and duplicate options are dropped.

import daft
from daft.functions import jev_choice

df = daft.from_pydict({
    "ticket": [
        "I was billed twice for my subscription this month.",
        "The app crashes every time I try to log in.",
        "Someone logged into my account from another country.",
        "Do you offer discounts for teams of 50 or more?",
    ],
})

df = df.with_column(
    "team",
    jev_choice(
        daft.col("ticket"),
        "which team should handle this?",
        ["billing", "technical", "security", "sales"],
    ),
)

df.show()

Scoring with `jev_score` Copy Chapter

The jev_score function places each row on an ordered rubric of 2 to 10 levels, listed lowest first. The result is the index of a level, weighted by the model's probability for each level, so it can fall between levels: with the levels below, a score of 2.5 sits between premium and luxury.

import daft
from daft.functions import jev_score

df = daft.from_pydict({
    "product": [
        "Plastic garden stool",
        "Solid oak dining chair",
        "Hand-stitched Italian leather armchair",
    ],
})

df = df.with_column(
    "tier",
    jev_score(
        daft.col("product"),
        "how luxurious is this product?",
        ["budget", "mid-range", "premium", "luxury"],
    ),
)

df.sort("tier", desc=True).show()

!!! note "Nulls and Batching"
Null rows, including NaN, are never sent to the model and produce a null result. The remaining rows are sent 25 per request with up to 16 requests in flight, and requests that time out, are rate-limited, or fail on the server are retried.

Ai Functions Prompt (`docs/ai-functions/prompt.md`) Copy Section

Prompting Language Models with Daft

A key strength of Daft is its ability to parallelize data processing on structured data. The prompt function is a powerful tool for use-cases like knowledge extraction, synthetic data generation, and batch tool-calling.

• Prompt Basics

  Learn how to compose prompt templates using DataFrame syntax, configure models, and pass optional parameters like temperature and max_tokens.

• Structured Outputs

  Generate structured data from language models using Pydantic models that automatically convert to Daft DataTypes with native struct support.

• Multimodal Inputs (Text, Images, Files)

  Pass text, images, and files (PDF, Markdown, HTML, CSV) as multimodal inputs to language models with automatic content formatting.

• Tool Calling

  Use OpenAI's built-in tools (web search, file search, code interpreter) or define custom functions for agentic workflows at scale.

!!! note
prompt is under active development, so make sure to stay up to date with our latest releases on github or through announcements in our slack community.

Quick Start Copy Chapter

The prompt function enables you to call language models on DataFrame columns, making it easy to scale LLM operations across thousands or millions of rows with automatic batching and parallelization.

The default provider for prompt is OpenAI. This means you can use OpenAI's latest models without specifying a provider. Just make sure your OPENAI_API_KEY environment variable is set and an OpenAIProvider will be created and attached to your daft.Session for you.

import daft
from daft.functions import prompt

# read some sample data
df = daft.from_pydict({
    "input": [
        "What sound does a cat make?",
        "What sound does a dog make?",
        "What sound does a cow make?",
        "What does the fox say?",
    ]
})

# generate responses using a chat model
df = df.with_column("response", prompt(daft.col("input"), model="gpt-5.4-mini"))

df.show()

╭─────────────────────────────┬─────────────────────────────╮
│ input                       ┆ response                    │
│ ---                         ┆ ---                         │
│ String                      ┆ String                      │
╞═════════════════════════════╪═════════════════════════════╡
│ What sound does a cat make? ┆ A cat makes a "meow" sound. │
│ What sound does a dog make? ┆ A dog makes a "woof" sound. │
│ What sound does a cow make? ┆ A cow makes a "moo" sound.  │
│ What does the fox say?      ┆ The fox says "ring-ding-... │
╰─────────────────────────────┴─────────────────────────────╯

Supported Providers for `prompt` Copy Section

1. OpenAI (Default) - Leverage models from OpenAI or any OpenAI-compatible provider.
2. Google - Take advantage of Google's latests models like Gemini 3 Pro.
3. vLLM Prefix Caching (Beta) - Distributed batch inference on vLLM built by the Daft Cloud team.
4. Transformers - Run open-weight instruction-tuned models locally via HuggingFace Transformers.

Prompt Basics Copy Chapter

The prompt function supports a wide range of tasks, from simple text generation to complex agentic workflows with tool calling. Daft's native multimodal data handling and scalable execution make it easy to process large datasets efficiently. This section covers core patterns for working with prompts, including templates, parameters, and model configuration.

!!! note
The default endpoint for OpenAI providers is v1/responses. You can pass use_chat_completions=True as an option to use the Chat Completions API instead of the new Responses API.

Working with OpenAI-compatible providers Copy Section

This example shows how to use the prompt function with an OpenAI-compatible provider like OpenRouter to classify anime quotes.

import os
import daft
from daft.functions import prompt

daft.set_provider(
    "openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ.get("OPENROUTER_API_KEY")
)

df = daft.from_pydict({
    "quote": [
        "I am going to be the king of the pirates!",
        "I'm going to be the next Hokage!",
    ],
})

df = (
    df
    .with_column(
        "response",
        prompt(
            daft.col("quote"),
            system_message="Classify the anime from the quote and return the show, character name, and explanation.",
            provider="openai",
            model="nvidia/nemotron-nano-9b-v2:free"
        )
    )
)

df.show(format="fancy", max_width=120)

╭───────────────────────────────────────────┬──────────────────────────────────────────────────────────────╮
│ quote                                     ┆ response                                                     │
╞═══════════════════════════════════════════╪══════════════════════════════════════════════════════════════╡
│ I am going to be the king of the pirates! ┆ **Show:** *One Piece*                                        │
│                                           ┆ **Character Name:** Monkey D. Luffy                          │
│                                           ┆ **Explanation:**                                             │
│                                           ┆ The quote "I am going to be the king…                        │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ I'm going to be the next Hokage!          ┆ **Show:** *Naruto*                                           │
│                                           ┆ **Character Name:** Naruto Uzumaki                           │
│                                           ┆ **Explanation:** The quote "I'm going to be the next Hokage… │
╰───────────────────────────────────────────┴──────────────────────────────────────────────────────────────╯
(Showing first 2 of 2 rows)

For more details on how to use OpenAI-compatible providers with the prompt function like OpenRouter, Hugging Face Inference Providers, Databricks, and more, check out the Prompting OpenAI-Compatible Providers section. Additionally, check out the Prompting with vLLM Online Serving section for more details on how to use vLLM Online Serving.

Running models locally with the Transformers provider Copy Section

The transformers provider runs open-weight instruction-tuned models locally via HuggingFace pipeline("text-generation"). Useful for offline batch inference without standing up a separate inference service.

pip install 'daft[transformers]'

import daft
from daft.functions import prompt

df = daft.from_pydict({"quote": ["I am going to be the king of the pirates!"]})

df = df.with_column(
    "response",
    prompt(
        daft.col("quote"),
        provider="transformers",
        model="HuggingFaceTB/SmolLM2-360M-Instruct",
        system_message="Classify the anime from the quote in one short sentence.",
        max_new_tokens=100,
    ),
)
df.show()

Generation kwargs (max_new_tokens, temperature, top_p, ...) are forwarded to the pipeline call. Constructor-time options (dtype, device_map, trust_remote_code, device, ...) go through pipeline_kwargs. The provider auto-selects cuda > mps > cpu.

You can also pass text-file columns (daft.File or raw bytes with a text/* MIME type) directly. The file contents are read and inlined into the prompt wrapped in a <file_*>...</file_*> tag derived from the MIME type, so the model can distinguish file content from your instructions:

df = daft.from_files("s3://my-bucket/docs/*.md")
df = df.with_column(
    "summary",
    prompt(
        [daft.col("file"), "Summarize this document in 3 bullets."],
        provider="transformers",
        model="HuggingFaceTB/SmolLM2-360M-Instruct",
    ),
)

Building Prompt Templates with the `format` function Copy Section

In addition to passing plain strings, you can build dynamic prompt templates using Daft's format function. This works similarly to Python's format, but works with Daft expressions. You can also define prompt templates with user-defined functions. With @daft.func you can pass in expressions or plain python strings for more flexible composition.

import daft
from daft.functions import prompt, format

# Define a prompt template with "format"
def answer_in_another_language(language: str, column_name: str) -> daft.Expression:
    return format(
        "Answer the following text in {}: {}",
        daft.lit(language),
        daft.col(column_name),
    )

# Alternatively define a row-wise UDF
@daft.func
def answer_in_another_language_func(language: str, question: str) -> str:
    return f"Answer the following text in {language}: {question}"

# Define some toy data
df = daft.from_pydict({
    "text": [
        "In what state(s) does water reach 101 degrees celsius?",
        "What are the total number of planets in our solar system?",
        "Who was the first person to discover radioactivity?",
    ]
})

# Lazily plan to create two new columns
df = (
    df
    .with_column(
        "answer1",
        prompt(
            answer_in_another_language("Chinese", "text"),
            model="google/gemma-3-4b",
            provider="lm_studio"  # Make sure you have google/gemma-3-4b loaded & running in lm_studio (local)
        )
    )
    .with_column(
        "answer2",
        prompt(
            answer_in_another_language_func(daft.lit("Spanish"), daft.col("text")),
            model="google/gemma-3-4b",
            provider="lm_studio"
        )
    )
)

df.show(format="fancy", max_width=80)

Optional parameters Copy Section

Passing additional arguments like temperature or max_tokens can simply be added to any prompt call as a keyword argument. All **kwargs are routed to the inference request unless documented otherwise.

Instructions for the model can also be added via the system_message argument which automatically prepends user content messages:

{"role": "system", "content": "You are an expert Chinese translation interpreter"}

For instance, if we wanted to call OpenAI via the Chat Completions endpoint, specifying temperature and max_tokens we would define a script like:

import daft
from daft.functions import prompt

df = daft.from_pydict({
    "prompt": [
        "What is the triple point of water?",
    ]
})

# Prompt Usage for GPT-4.1 Chat Completions
df = df.with_column(
    "result",
    prompt(
        messages=daft.col("prompt"),
        system_message="You are a helpful assistant.",
        model="gpt-4.1-2025-04-14",
        provider="openai",
        temperature=0.717,  # Specify temperature
        max_tokens=200,     # Specify max_tokens
        use_chat_completions=True,
    )
)

However, the Responses API uses different parameter names. For GPT-5 models, use max_output_tokens instead of max_tokens, and note that temperature is not supported:

import daft
from daft.functions import prompt

df = daft.from_pydict({
    "prompt": [
        "What is the triple point of water?",
    ]
})

# Prompt Usage for GPT-5 Responses
df = df.with_column(
    "result",
    prompt(
        messages=daft.col("prompt"),
        system_message="You are a helpful assistant.",
        model="gpt-5",
        provider="openai",
        max_output_tokens=200,  # Specify max_output_tokens
    )
)

Reasoning Models Copy Section

GPT-5 and other reasoning-capable models support the reasoning parameter to control compute allocation. Pass it as you would in a client.responses.create() request:

import daft
from daft.functions import prompt

df = daft.from_pydict({
    "prompt": [
        "Find me buy-one-get-one-free burrito deals in SF right now.",
    ]
})

# Prompt Usage for GPT-5 Responses
df = df.with_column(
    "result",
    prompt(
        messages=daft.col("prompt"),
        system_message="You are a helpful assistant.",
        model="gpt-5-2025-08-07",
        provider="openai",
        reasoning={"effort": "high"},  # Adjust reasoning level
    )
)

See also: Providers Overview.

Tools Copy Section

The easiest way to get started with tool calling is to use OpenAI's built-in tools like web-search.

import daft
from daft.functions import prompt
from datetime import datetime

df = daft.from_pydict({
    "query": [
        "Buy one get one free burritos in SF.",
    ]
})

df = df.with_column(
    "search_results",
    prompt(
        daft.col("query"),
        model="gpt-5",
        tools=[{"type": "web_search"}],
    )
)

df.write_csv(f"../.data/prompt/oai_web_search_{str(datetime.now())}")

Multimodal Inputs Copy Chapter

Daft is purpose-built for multimodal AI workloads. The prompt function accepts text, images, and files as a list of expressions in the messages parameter, making it easy to work with mixed content types.

import daft
from daft.functions import prompt

df = daft.from_pydict({
    "prompt": ["What's in this image and file?"],
    "my_image": ["/Users/MyName/Downloads/SigmundTheCat.png"],
    "my_file": ["/Users/MyName/Downloads/report.pdf"],
})

# Decode the image and file paths
df = df.with_column(
    "my_image",
    daft.functions.decode_image(daft.col("my_image").download())
)

df = df.with_column(
    "my_file",
    daft.functions.file(daft.col("my_file"))
)

# Prompt Usage for GPT-5 Responses
df = df.with_column(
    "result",
    prompt(
        messages=[daft.col("prompt"), daft.col("my_image"), daft.col("my_file")],
        system_message="You are a helpful assistant.",
        model="gpt-5-2025-08-07",
        provider="openai",
        reasoning={"effort": "high"},  # Adjust reasoning level
        tools=[{"type": "web_search"}],  # Leverage internal OpenAI Tools
    )
)

Working with Files Copy Section

The prompt function supports multiple file input methods depending on the provider and file type:

• PDF files: Passed directly as file inputs (native OpenAI support)
• Text files (Markdown, HTML, CSV, etc.): Content is automatically extracted and injected into prompts
• Images: Supported via daft.DataType.Image() or file paths

PDF Files

To include PDF files, create a daft.File column and pass it in your messages list:

import daft
from daft import lit, col
from daft.functions import prompt, file, format
from dotenv import load_dotenv

load_dotenv()

# Discover PDF files from HuggingFace
df = daft.from_glob_path("hf://datasets/Eventual-Inc/sample-files/papers/*.pdf")

df = (
    df
    # Create a daft.File column from the path
    .with_column("file", file(col("path")))
    # Prompt GPT-5-nano with PDF files as context
    .with_column(
        "response",
        prompt(
            [lit("What are in the contents of this file?\n"), col("file")],
            model="gpt-5-nano",
            provider="openai",
        )
    )
)
df.show(format="fancy", max_width=80)

/* Detailed source-code truncated for AI context efficiency. */

Text Files (Markdown, HTML, CSV)

While OpenAI's native API only supports PDFs directly, attempting to pass other file types like Markdown, CSV, or HTML will throw errors like:

Invalid input: Expected file type to be a supported format: .pdf but got .jsonl”

To work around this, Daft automatically extracts content from any text/* MIME type file and injects it into the prompt with XML-wrapped tags. For example, to use Markdown files as context:

import daft
from daft import lit, col
from daft.functions import prompt, file, format

# Discover Markdown Files in your Documents Folder
df = daft.from_glob_path("/Users/MyName/Documents/*.md")

df = (
    df
    # Create a daft.File column from the path
    .with_column("file", file(col("path")))
    # Prompt GPT-5-nano with markdown files as context
    .with_column(
        "response",
        prompt(
            messages=[lit("What are in the contents of this file?\n"), col("file")],
            model="gpt-5-nano",
            provider="openai",
        )
    )
)
df.show(format="fancy", max_width=80)

When you pass [lit("What are in the contents of this file?\n"), col("file")], Daft automatically formats the content. The resulting prompt will look like:

What are in the contents of this file?
<file_text_markdown>
## Some Title
Example Markdown content with a [link](https://somelink.com).
</file_text_markdown>

Structured Outputs Copy Chapter

Daft natively supports converting Pydantic Models to Daft Datatypes. This simplifies structured generation in a vectorized context where nested fields materialize as structs.

import os
from dotenv import load_dotenv

import daft
from daft.functions import prompt, unnest
from pydantic import BaseModel, Field


load_dotenv()

class Anime(BaseModel):
    show: str = Field(description="The name of the anime show")
    character: str = Field(description="The name of the character who says the quote")
    explanation: str = Field(description="Why the character says the quote")

daft.set_provider(
    "openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ.get("OPENROUTER_API_KEY"),
)

# Create a dataframe with the quotes
df = daft.from_pydict(
    {
        "quote": [
            "I am going to be the king of the pirates!",
            "I'm going to be the next Hokage!",
        ],
    }
)

# Use the prompt function to classify the quotes
df = df.with_column(
    "nemotron-response",
    prompt(
        daft.col("quote"),
        system_message="You are an anime expert. Classify the anime based on the text and return the name, character, and quote.",
        return_format=Anime,
        provider="openai",
        model="nvidia/nemotron-nano-9b-v2:free",
    ),
).select("quote", unnest(daft.col("nemotron-response")))

df.show(format="fancy", max_width=120)

Structured Outputs with vLLM OpenAI Online Server Copy Section

Sometimes Pydantic structured outputs is overkill. vLLM's structured outputs syntax uses the extra_body param which we can use thanks to how Daft wires kwargs to the model request. Keep in mind the normal Pydantic structured outputs via return_format also works with vLLM online serving.

One of the fastest ways to get started with online serving with vLLM is with Google Colab. To launch an OpenAI compatible server on vLLM run the following command in your terminal on an L4 GPU instance or larger:

python -m vllm.entrypoints.openai.api_server \
  --model google/gemma-3-4b-it \
  --guided-decoding-backend guidance \
  --dtype bfloat16 \
  --gpu-memory-utilization 0.85 \
  --host 0.0.0.0 --port 8000

If you are in Google Colab, you can open a terminal by clicking the terminal icon in the bottom left of the ui. It usually takes at least 7.5 minutes before the vLLM server is ready.

Setting up vLLM under the OpenAI provider

import daft
from daft.functions import prompt, format

daft.set_provider(
    "openai",
    api_key="none",
    base_url="http://localhost:8000/v1",
)

Guided Choice (Classification)

df = daft.from_pydict({
    "statement": [
        "vLLM is wonderful",
        "Daft is wicked fast",
        "Slow inference sucks",
    ]
})

df = df.with_column(
    "sentiment",
    prompt(
        messages=format("Classify this sentiment: {}", daft.col("statement")),
        model="google/gemma-3-4b-it",
        extra_body={"structured_outputs": {"choice": ["positive", "negative"]}},
    )
)

Regex

df = daft.from_pydict({
    "name": [
        "Alan Turing",
        "Steve Jobs",
        "Sammy Sidhu",
        "Jay Chia",
    ],
    "company": [
        "enigma.uk",
        "apple.com",
        "daft.ai",
        "daft.ai",
    ]
})

df = df.with_column(
    "email",
    prompt(
        messages=format(
            "Generate an example email address for {}, who works at {}. Example result: [email protected]\nMake sure to finish with a new line symbol\n",
            daft.col("name"),
            daft.col("company")
        ),
        model="google/gemma-3-4b-it",
        extra_body={"structured_outputs": {"regex": r"\w+@\w+\.com\n"}, "stop": ["\n"]},
    )
)

Ai Functions Providers (`docs/ai-functions/providers.md`) Copy Section

AI Providers Overview

Providers in Daft give you flexible control over where and how your AI inference runs. Whether you're using commercial APIs like OpenAI, running local models with LM Studio, or leveraging open-source models from Hugging Face, providers offer a unified interface for configuration.

This guide covers:

• Quick provider setup
• Setting named providers within sessions
• Working with multiple providers
• Prompting with OpenAI-Compatible Providers
• Prompting with vLLM Online Serving

Quick provider setup Copy Chapter

The simplest way to use a provider is with daft.set_provider(), which configures a provider globally for your Daft session:

import os
import daft

# Set OpenAI Provider for Global Daft Session
daft.set_provider(
    "openai",
    api_key=os.environ.get("OPENAI_API_KEY"),
    timeout=30.0,
    max_retries=3
)

# Retrieve Provider for an AI Function
provider = daft.get_provider("openai")

Supported Providers Copy Chapter

Daft supports the following AI Providers:

• OpenAI - API for GPT models and other AI capabilities
• Transformers - Hugging Face models for local inference
• Google - Google's Gemini API
• LM Studio - Run local models privately on your computer

Setting a named OpenAI Provider within a Session Copy Chapter

For more control, you can create named providers within a session. This is useful when working with OpenAI-compatible APIs like OpenRouter, or when you need to switch between different configurations.

You can pass a Provider instance directly to set_provider(), which will automatically attach it to the session:

import os
import daft
from daft.ai.openai.provider import OpenAIProvider

sess = daft.Session()
openrouter_provider = OpenAIProvider(
    name="OpenRouter",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ.get("OPENROUTER_API_KEY")
)
sess.set_provider(openrouter_provider)  # Auto-attaches and sets as current

# Retrieve Provider for an AI Function
provider = sess.get_provider("OpenRouter")

This is equivalent to the two-step approach of calling attach_provider() followed by set_provider():

# The two-step approach (still supported):
sess.attach_provider(openrouter_provider)
sess.set_provider("OpenRouter")

Using the Google Provider Copy Chapter

The Google provider enables you to use Google's Gemini models for text generation, multimodal processing, and structured outputs.

import daft
import os

# Set up the Google provider with your API key
with daft.session() as session:
    session.set_provider("google", api_key=os.environ["GOOGLE_API_KEY"])

    # Create a DataFrame with questions
    df = daft.from_pydict({
        "question": [
            "What is the capital of France?",
            "Explain quantum computing in simple terms.",
            "What are the benefits of fusion energy?"
        ]
    })

    # Use the prompt function with Google's Gemini model
    df = df.with_column(
        "answer",
        daft.functions.prompt(
            daft.col("question"),
            provider="google",
            model="gemini-2.5-flash"  # or "gemini-3-pro-preview"
        )
    )

    df.show()

The Google provider supports all the same features as other providers:

• Structured outputs with Pydantic models
• Multimodal inputs (text, images, PDFs, documents)
• System messages for custom instructions
• Custom generation config (temperature, max_output_tokens, etc.)

End-to-End Usage with Multiple Providers Copy Chapter

For complex workflows, you might need to use different providers for different tasks—for example, using GPT-5 for validation while using a cheaper model for initial classification. Daft makes it easy to manage multiple providers in a single session.

/* Detailed source-code truncated for AI context efficiency. */

This example demonstrates:

• Multiple API keys: Different providers for development and production environments
• Cost optimization: Use cheaper models for initial processing, premium models for validation
• Provider routing: Explicitly specify which provider to use for each operation

!!! tip "When to Use Multiple Providers"

Multiple providers are useful when you need to:

- **Balance cost and quality**: Use cheaper models for bulk processing, premium models for critical tasks
- **Ensure redundancy**: Fall back to alternative providers if one fails
- **Compare model outputs**: Run the same prompt through different models for quality assessment
- **Manage rate limits**: Distribute load across multiple API keys or providers

Prompting with OpenAI-Compatible Providers Copy Chapter

You can use OpenAI-compatible providers with the prompt function like OpenRouter, HuggingFace Inference Providers, Databricks, and more.

import os
import daft

# For OpenRouter
# See: https://openrouter.ai/docs/quickstart#using-the-openai-sdk
daft.set_provider(
    "openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY")
)

# For Hugging Face Inference Providers
# See: https://huggingface.co/inference/get-started
daft.set_provider(
    "openai",
    base_url="https://router.huggingface.co/v1",
    api_key=os.getenv("HF_TOKEN")
)

# For Databricks
# How to get your Databricks token: https://docs.databricks.com/en/dev-tools/auth/pat.html
# More info on model serving: https://docs.databricks.com/aws/en/machine-learning/model-serving/score-foundation-models
daft.set_provider(
    "openai",
    base_url="https://<workspace-id>.cloud.databricks.com/serving-endpoints",
    api_key=os.getenv("DATABRICKS_TOKEN"),
)

Prompting with vLLM Online Serving Copy Chapter

For vLLM Online Serving, you can set the provider as demonstrated in the example below.

Note: vLLM's OpenAI-compatible endpoint expects requests in the Chat Completions API format. Daft's prompt() function defaults to using the newer Responses API format, which is not supported by vLLM. To ensure compatibility, set use_chat_completions=True in prompt() so that your requests use the Chat Completions API format required by vLLM.

import os
import daft

# For vLLM Online Serving
daft.set_provider(
    "openai",
    api_key="none",
    base_url="http://localhost:8000/v1",
) # Make sure to set use_chat_completions=True in prompt()

df = df.with_column(
    "response",
    prompt(
        daft.col("input"),
        use_chat_completions=True,
        model="google/gemma-3-4b-it",
    )
)

df.show()

The following sample cli launch command is optimized for Google Colab's A100 High-RAM instance and Google's gemma-3-4b-it model.

python -m vllm.entrypoints.openai.api_server \
--model google/gemma-3-4b-it \
--enable-chunked-prefill \
--guided-decoding-backend guidance \
--dtype bfloat16 \
--gpu-memory-utilization 0.85 \
--host 0.0.0.0 --port 8000

• Server readiness may take ~7–8 minutes
• For vLLM online serving, set api_key = "none" and base_url = "http://0.0.0.0:8000/v1"
• guided-decoding-backend guidance is required for structured outputs.

More Resources Copy Chapter

• AI Functions Overview
• Embedding Text and Images
• Classify Text and Images
• Contributing AI Functions

Api Aggregations (`docs/api/aggregations.md`) Copy Section

Aggregations

When performing aggregations such as sum, mean and count, Daft enables you to group data by certain keys and aggregate within those keys.

Calling [df.groupby()][daft.DataFrame.groupby] returns a GroupedDataFrame object which is a view of the original DataFrame but with additional context on which keys to group on. You can then call various aggregation methods to run the aggregation within each group, returning a new DataFrame.

::: daft.dataframe.dataframe.GroupedDataFrame
options:
filters: ["!^_"]

Api Ai (`docs/api/ai.md`) Copy Section

AI

Feature coverage table:

!!! note
Open an issue on our GitHub if you would like to see support for a feature.

Functions Copy Chapter

::: daft.functions.ai.prompt
options:
heading_level: 3

::: daft.functions.ai.embed_text
options:
heading_level: 3

::: daft.functions.ai.embed_image
options:
heading_level: 3

::: daft.functions.ai.classify_text
options:
heading_level: 3

::: daft.functions.ai.classify_image
options:
heading_level: 3

Model Protocols Copy Chapter

::: daft.ai.protocols.Prompter
options:
heading_level: 3

::: daft.ai.protocols.TextEmbedder
options:
heading_level: 3

::: daft.ai.protocols.TextEmbedderDescriptor
options:
heading_level: 3

::: daft.ai.protocols.ImageEmbedder
options:
heading_level: 3

::: daft.ai.protocols.ImageEmbedderDescriptor
options:
heading_level: 3

::: daft.ai.protocols.TextClassifier
options:
heading_level: 3

::: daft.ai.protocols.TextClassifierDescriptor
options:
heading_level: 3

Providers Copy Chapter

::: daft.ai.provider.Provider
options:
filters: ["!^_"]
heading_level: 3

::: daft.ai.provider.load_provider
options:
heading_level: 3

::: daft.ai.provider.load_google
options:
heading_level: 3

::: daft.ai.provider.load_lm_studio
options:
heading_level: 3

::: daft.ai.provider.load_openai
options:
heading_level: 3

::: daft.ai.provider.load_transformers
options:
heading_level: 3

::: daft.ai.provider.load_vllm_prefix_caching
options:
heading_level: 3

Api Catalogs Tables (`docs/api/catalogs_tables.md`) Copy Section

Catalogs and Tables

Daft integrates with various catalog implementations using its Catalog and Table interfaces. These are high-level APIs to manage catalog objects (tables and namespaces), while also making it easy to leverage Daft's existing daft.read_ and df.write_ APIs for open table formats like Iceberg and Delta Lake.

::: daft.catalog.Catalog
options:
filters: ["!^_"]

::: daft.catalog.Identifier
options:
filters: ["!^_"]

::: daft.catalog.Table
options:
filters: ["!^_"]

Api Config (`docs/api/config.md`) Copy Section

Configuration

Configure the execution backend, Daft in various ways during execution, and how Daft interacts with storage.

Setting the Runner Copy Chapter

Control the execution backend that Daft will run on by calling these functions once at the start of your application.

::: daft.set_runner_native
options:
heading_level: 3

::: daft.set_runner_ray
options:
heading_level: 3

::: daft.get_or_create_runner
options:
heading_level: 3

Checking the Runner Copy Chapter

Check the execution backend that Daft is currently using.

::: daft.get_or_infer_runner_type
options:
heading_level: 3

Setting Configurations Copy Chapter

Configure Daft in various ways during execution.

::: daft.context.set_planning_config
options:
heading_level: 3

::: daft.context.planning_config_ctx
options:
heading_level: 3

::: daft.context.set_execution_config
options:
heading_level: 3

::: daft.context.execution_config_ctx
options:
heading_level: 3

I/O Configurations Copy Chapter

Configure behavior when Daft interacts with storage (e.g. credentials, retry policies and various other knobs to control performance/resource usage)

These configurations are most often used as inputs to Daft when reading I/O functions such as in I/O.

::: daft.daft.IOConfig
options:
filters: ["!^_"]

::: daft.io.S3Config
options:
filters: ["!^_"]

::: daft.io.S3Credentials
options:
filters: ["!^_"]

::: daft.io.GCSConfig
options:
filters: ["!^_"]

::: daft.io.AzureConfig
options:
filters: ["!^_"]

::: daft.io.HTTPConfig
options:
filters: ["!^_"]

::: daft.io.UnityConfig
options:
filters: ["!^_"]

::: daft.io.HuggingFaceConfig
options:
filters: ["!^_"]

::: daft.io.TosConfig
options:
filters: ["!^_"]

::: daft.io.CosConfig
options:
filters: ["!^_"]

Api Dataframe (`docs/api/dataframe.md`) Copy Section

DataFrame

Most DataFrame methods are lazy, meaning that they do not execute computation immediately when invoked. Instead, these operations are enqueued in the DataFrame's internal query plan, and are only executed when Execution DataFrame methods are called.

::: daft.DataFrame
options:
filters: ["!^[^]", "!repr", "!__builder"]

Api Datasets (`docs/api/datasets.md`) Copy Section

Datasets

Daft provides simple, performant, and responsible ways to access useful datasets like Common Crawl and DROID.

Common Crawl Copy Chapter

Check out our Common Crawl dataset guide for more examples!

::: daft.datasets.common_crawl.common_crawl
options:
filters: ["!^_"]
heading_level: 3

LeRobot Copy Chapter

See the LeRobot dataset guide for v2.1 / v3 layouts, episode vs frame workflows, and Hub/local paths.

::: daft.datasets.lerobot.read
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.lerobot.read_episodes
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.lerobot.load_episode_frames
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.lerobot.read_tasks

DROID Copy Chapter

Check out our DROID dataset guide for more examples!

::: daft.datasets.droid.raw
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.droid.scenes
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.droid.trajectory
options:
filters: ["!^_"]
heading_level: 3

::: daft.datasets.droid.camera_frames
options:
filters: ["!^_"]
heading_level: 3

Api Expressions (`docs/api/expressions.md`) Copy Section

Expressions

Daft Expressions allow you to express some computation that needs to happen in a DataFrame. This page provides an overview of all the functionality that is provided by Daft Expressions.

Constructors Copy Chapter

::: daft.expressions.col
options:
heading_level: 3

::: daft.expressions.lit
options:
heading_level: 3

::: daft.sql.sql.sql_expr
options:
heading_level: 3

::: daft.expressions.Expression
options:
filters: ["!^[^]", "!over", "!lag", "!lead"]

::: daft.expressions.WhenExpr
options:
filters: ["!^_"]

::: daft.expressions.visitor.ExpressionVisitor
options:
filters: ["!^_"]

Api Io (`docs/api/io.md`) Copy Section

I/O

Daft offers a variety of approaches to creating a DataFrame from reading various data sources (in-memory data, files, data catalogs, and integrations) and writing to various data sources. See more about other Connectors in Daft User Guide.

Input Copy Chapter

::: daft.from_arrow
options:
heading_level: 3

::: daft.from_dask_dataframe
options:
heading_level: 3

::: daft.from_glob_path
options:
heading_level: 3

::: daft.from_pandas
options:
heading_level: 3

::: daft.from_pydict
options:
heading_level: 3

::: daft.from_pylist
options:
heading_level: 3

::: daft.from_ray_dataset
options:
heading_level: 3

::: daft.read_blob
options:
heading_level: 3

::: daft.read_csv
options:
heading_level: 3

::: daft.read_deltalake
options:
heading_level: 3

::: daft.read_hudi
options:
heading_level: 3

::: daft.read_iceberg
options:
heading_level: 3

::: daft.read_json
options:
heading_level: 3

::: daft.read_kafka
options:
heading_level: 3

::: daft.read_lance
options:
heading_level: 3

::: daft.read_parquet
options:
heading_level: 3

::: daft.read_orc
options:
heading_level: 3

::: daft.read_sql
options:
heading_level: 3

::: daft.read_video_frames
options:
heading_level: 3

::: daft.read_warc
options:
heading_level: 3

::: daft.read_webdataset
options:
heading_level: 3

::: daft.read_huggingface
options:
heading_level: 3

::: daft.sql.sql.sql
options:
heading_level: 3

Output Copy Chapter

::: daft.dataframe.DataFrame.write_csv
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_json
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_parquet
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_deltalake
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_iceberg
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_lance
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_sql
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_bigtable
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_clickhouse
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_huggingface
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_turbopuffer
options:
heading_level: 3

::: daft.dataframe.DataFrame.write_sink
options:
heading_level: 3

User-Defined Copy Chapter

Daft supports diverse input sources and output sinks, this section covers lower-level APIs which we are evolving for more advanced usage.

!!! warning "Warning"

These APIs are considered experimental.

::: daft.io.source.DataSource
options:
filters: ["!^_"]
heading_level: 3

::: daft.io.source.DataSourceTask
options:
filters: ["!^_"]
heading_level: 3

::: daft.io.sink.DataSink
options:
filters: ["!^_"]
heading_level: 3

::: daft.io.sink.WriteResult
options:
filters: ["!^_"]
heading_level: 3

Pushdowns Copy Chapter

Daft supports predicate, projection, and limit pushdowns.

::: daft.io.pushdowns.Pushdowns
options:
filters: ["!^_"]
heading_level: 3

::: daft.io.scan.ScanOperator
options:
filters: ["!^_"]

Api Misc (`docs/api/misc.md`) Copy Section

Miscellaneous

Image Types Copy Chapter

::: daft.daft.ImageMode
options:
filters: ["!^_"]

::: daft.daft.ImageFormat
options:
filters: ["!^_"]

Api Schema (`docs/api/schema.md`) Copy Section

Schema

Daft can display your DataFrame's schema without materializing it. Under the hood, it performs intelligent sampling of your data to determine the appropriate schema, and if you make any modifications to your DataFrame it can infer the resulting types based on the operation.

::: daft.schema.Schema
options:
filters: ["!^_"]

!!! warning ""

Schema has been moved to `daft.schema` but is still accessible at `daft.logical.schema`.

::: daft.logical.schema.Schema
options:
filters: ["!^_"]

Api Series (`docs/api/series.md`) Copy Section

Series

Each column in a Table is a Series. Series expose methods which invoke high-performance kernels for manipulation of a column of data.

::: daft.series.Series
options:
filters: ["^to", "^from"]

Api Sessions (`docs/api/sessions.md`) Copy Section

Sessions

Sessions enable you to attach objects such as catalogs, providers, models, functions, and tables which are reference-able in DataFrame operations. The session also enables creating temporary objects which are accessible through both the Python and SQL APIs. Sessions hold configuration state such as current_catalog and current_namespace which are used in name resolution and can simplify your workflows. Learn more about Sessions in the Daft User Guide.

::: daft.session.session
options:
heading_level: 3

::: daft.session.current_session
options:
heading_level: 3

::: daft.session.Session
options:
filters: ["!^_"]

Api Udf (`docs/api/udf.md`) Copy Section

User-Defined Functions

User-Defined Functions (UDFs) are a mechanism to run Python code on the data that lives in a DataFrame. A UDF can be used just like Expressions, allowing users to express computation that should be executed by Daft lazily.

To write a UDF, use the @daft.func or @daft.cls decorators to turn Python functions into Daft operations that work seamlessly with DataFrame expressions. Daft also supports custom aggregation functions, or UDAFs with @daft.udaf with a three-stage pipeline (aggregate, combine, finalize) that plugs into Daft's distributed aggregation engine.

Learn more about user-defined functions in the user guide: daft.func, daft.cls, daft.udaf.

Stateless Function UDFs Copy Chapter

::: daft.func

::: daft.udf._FuncDecorator

Stateful Class UDFs Copy Chapter

::: daft.cls

::: daft.method

::: daft.udf._MethodDecorator

::: daft.udf.udf_v2.Func

Aggregate UDFs Copy Chapter

::: daft.udaf

Api Window (`docs/api/window.md`) Copy Section

Window Functions

Window functions allow you to perform calculations across a set of rows that are related to the current row. They operate on a group of rows (called a window frame) and return a result for each row based on the values in its window frame, without collapsing the result into a single row like aggregate functions do. See Window Functions Tutorial for a step-by-step tutorial.

::: daft.window.Window

Applying Window Functions Copy Chapter

::: daft.expressions.Expression.over
options:
heading_level: 3

Aggregate Functions Copy Chapter

Standard aggregate functions (e.g., [sum][daft.expressions.Expression.sum], [mean][daft.expressions.Expression.mean], [count][daft.expressions.Expression.count], [min][daft.expressions.Expression.min], [max][daft.expressions.Expression.max], etc) can be used as window functions by applying them with [.over][daft.expressions.Expression.over]. They work with all valid window specifications (partition by only, partition + order by, partition + order by + frame). Refer to the Expressions API for a full list of aggregate functions.

!!! note "Note"
When using aggregate functions with both partition by and order by, the default window frame includes all rows from the start of the partition up to the current row — equivalent to rows between unbounded preceding and current row.

Ranking Functions Copy Chapter

These functions compute ranks within a window partition. They require an [order_by][daft.window.Window.order_by] clause without a [rows_between][daft.window.Window.rows_between] or [range_between][daft.window.Window.range_between] clause in the window specification.

• daft.functions.row_number
• daft.functions.rank
• daft.functions.dense_rank

Lead/Lag Functions Copy Chapter

These functions access data from preceding or succeeding rows within a window partition. They require an [order_by][daft.window.Window.order_by] clause without a [rows_between][daft.window.Window.rows_between] or [range_between][daft.window.Window.range_between] clause in the window specification.

::: daft.expressions.Expression.lag
options:
heading_level: 3

::: daft.expressions.Expression.lead
options:
heading_level: 3

Configuration Sessions Usage (`docs/configuration/sessions-usage.md`) Copy Section

Daft Sessions, Catalogs, and Tables

Sessions enable you to attach catalogs, tables, and create temporary objects which are accessible through both the Python and SQL APIs. Sessions hold configuration state such as current_catalog and current_namespace which are used in name resolution and can simplify your workflows.

Example Copy Chapter

import daft

# `import daft` defines an implicit session `daft.current_session()`

from daft.session import Session

# create a new session
sess = Session()

# create a temp table from a DataFrame
sess.create_temp_table("T", daft.from_pydict({ "x": [1,2,3] }))

# read table as dataframe from the session
_ = sess.read_table("T")

# get the table instance
t = sess.get_table("T")

# read table instance as a datadrame
_ = t.read()

# execute sql against the session
sess.sql("SELECT * FROM T").show()
╭───────╮
│ x     │
│ ---   │
│ Int64 │
╞═══════╡
│ 1     │
├╌╌╌╌╌╌╌┤
│ 2     │
├╌╌╌╌╌╌╌┤
│ 3     │
╰───────╯

Usage Copy Chapter

This section covers detailed usage of the current APIs with some code snippets.

Setup Copy Section

!!! note "Note"

For these examples, we are using sqlite Iceberg which requires `pyiceberg[sql-sqlite]`.

from daft import Catalog
from pyiceberg.catalog.sql import SqlCatalog

# don't forget to `mkdir -p /tmp/daft/example`
tmpdir = "/tmp/daft/example"

# create a pyiceberg catalog backed by sqlite
iceberg_catalog = SqlCatalog(
    "default",
    **{
        "uri": f"sqlite:///{tmpdir}/catalog.db",
        "warehouse": f"file://{tmpdir}",
    },
)

# creating a daft catalog from the pyiceberg catalog implementation
catalog = Catalog.from_iceberg(iceberg_catalog)

# check
catalog.name
"""
'default'
"""

Session State Copy Section

Let's get started by creating an empty session and checking the state.

import daft

from daft.session import Session

# create a new empty session
sess = Session()

# check the current catalog (None)
sess.current_catalog()

# get the current namespace (None)
sess.current_namespace()

Attach & Detach Copy Section

The attach and detach methods make it easy to use catalogs and tables in a session. This example shows how we can attach our newly created catalog. When you attach a catalog to an empty session, it automatically becomes the current active catalog.

# attach makes it possible to use existing catalogs in the session
sess.attach(catalog)

# check the current catalog was set automatically
sess.current_catalog()
"""
Catalog('default')
"""

# detach would remove the catalog
# sess.detach_catalog("default")

Create & Drop Copy Section

We can create tables and namespaces directly through a catalog or via our session.

# create a namespace 'example'
sess.create_namespace("example")

# verify it was created
sess.list_namespaces()
"""
[Identifier('example')]
"""

# you can create an empty table with a schema
# sess.create_table("example.tbl", schema)

# but suppose we have some data..
df = daft.from_pydict({
    "x": [ True, True, False ],
    "y": [ 1, 2, 3 ],
    "z": [ "abc", "def", "ghi" ],
})

# create a table from the dataframe, which will create + append
sess.create_table("example.tbl", df)

# you can also create temporary tables from dataframes
# > echo "x,y,z\nFalse,4,jkl" > /tmp/daft/row.csv
sess.create_temp_table("temp", daft.read_csv("/tmp/daft/row.csv"))

# you can drop too
# sess.drop_table("example.tbl")
# sess.drop_namespace("example")

Read & Write Copy Section

Using sessions abstracts away underlying catalog and table implementations so you can easily read and write DataFrames.

/* Detailed source-code truncated for AI context efficiency. */

Using SQL Copy Section

The session enables executing Daft SQL against your catalogs.

# use the given catalog and namespace (like the use/set_ methods)
sess.sql("USE default.example")

# we support both qualified and unqualified names by leveraging the session state
sess.sql("SELECT * FROM tbl LIMIT 1").show()
╭─────────┬───────┬──────╮
│ x       ┆ y     ┆ z    │
│ ---     ┆ ---   ┆ ---  │
│ Boolean ┆ Int64 ┆ String │
╞═════════╪═══════╪══════╡
│ true    ┆ 4     ┆ jkl  │
╰─────────┴───────┴──────╯

# we can even combine our queries with the temp table from earlier
sess.sql("SELECT * FROM example.tbl, temp LIMIT 1").show()
╭─────────┬───────┬──────┬────────────┬────────┬────────╮
│ x       ┆ y     ┆ z    ┆      …     ┆ temp.y ┆ temp.z │
│ ---     ┆ ---   ┆ ---  ┆            ┆ ---    ┆ ---    │
│ Boolean ┆ Int64 ┆ String ┆ (1 hidden) ┆ Int64  ┆ String   │
╞═════════╪═══════╪══════╪════════════╪════════╪════════╡
│ true    ┆ 4     ┆ jkl  ┆ …          ┆ 4      ┆ jkl    │
╰─────────┴───────┴──────┴────────────┴────────┴────────╯

!!! note "Note"

We aim to support SQL DDL in future releases!

Reference Copy Chapter

For complete documentation, please see the Session API docs.

Connectors Aws (`docs/connectors/aws.md`) Copy Section

Reading from and Writing to AWS S3

Daft has record-setting performance when reading from and writing to AWS S3, and understands natively the URL protocol s3:// as referring to data that resides in S3.

Authorization/Authentication Copy Chapter

In AWS S3, data is stored under the hierarchy of:

1. Bucket: The container for data storage, which is the top-level namespace for data storage in S3.
2. Object Key: The unique identifier for a piece of data within a bucket.

URLs to data in S3 come in the form: s3://{BUCKET}/{OBJECT_KEY}.

Rely on Environment Copy Section

You can configure the AWS CLI to have Daft automatically discover credentials. Alternatively, you may specify your credentials in environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_SESSION_TOKEN.

Please be aware that when doing so in a distributed environment such as Ray, Daft will pick these credentials up from worker machines and thus each worker machine needs to be appropriately provisioned.

If instead you wish to have Daft use credentials from the "driver", you may wish to manually specify your credentials.

Manually specify credentials Copy Section

You may also choose to pass these values into your Daft I/O function calls using an [daft.io.S3Config][daft.io.S3Config] config object.

[daft.set_planning_config][daft.context.set_planning_config] is a convenient way to set your [daft.io.IOConfig][daft.io.IOConfig] as the default config to use on any subsequent Daft method calls.

=== "🐍 Python"

```python
from daft.io import IOConfig, S3Config

# Supply actual values: key_id is the AWS Access Key ID, access_key is the AWS Secret Access Key
io_config = IOConfig(s3=S3Config(key_id="key_id", session_token="session_token", access_key="access_key"))

# Globally set the default IOConfig for any subsequent I/O calls
daft.set_planning_config(default_io_config=io_config)

# Perform some I/O operation
df = daft.read_parquet("s3://my_bucket/my_path/**/*")
```

Alternatively, Daft supports overriding the default IOConfig per-operation by passing it into the io_config= keyword argument. This is extremely flexible as you can pass a different [daft.io.S3Config][daft.io.S3Config] per function call if you wish!

=== "🐍 Python"

```python
# Perform some I/O operation but override the IOConfig
df2 = daft.read_csv("s3://my_bucket/my_other_path/**/*", io_config=io_config)
```

Connectors Azure (`docs/connectors/azure.md`) Copy Section

Reading from and Writing to Azure Blob Store

Daft is able to read/write data to/from Azure Blob Store, and understands natively the URL protocols az:// and abfs:// as referring to data that resides in Azure Blob Store.

!!! warning "Warning"

Daft currently only supports globbing and listing files in storage accounts with [hierarchical namespaces](https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-namespace) enabled. Hierarchical namespaces enable Daft to use its embarrassingly parallel globbing algorithm to improve performance of listing large nested directories of data. Please file an [issue](https://github.com/Eventual-Inc/Daft/issues) if you need support for non-hierarchical namespace buckets! We'd love to support your use-case.

Authorization/Authentication Copy Chapter

In Azure Blob Service, data is stored under the hierarchy of:

1. Storage Account
2. Container (sometimes referred to as "bucket" in S3-based services)
3. Object Key

URLs to data in Azure Blob Store come in the form: az://{CONTAINER_NAME}/{OBJECT_KEY}.

Given that the Storage Account is not a part of the URL, you must provide this separately.

Rely on Environment Copy Section

You can rely on Azure's environment variables to have Daft automatically discover credentials.

Please be aware that when doing so in a distributed environment such as Ray, Daft will pick these credentials up from worker machines and thus each worker machine needs to be appropriately provisioned.

If instead you wish to have Daft use credentials from the "driver", you may wish to manually specify your credentials.

Manually specify credentials Copy Section

You may also choose to pass these values into your Daft I/O function calls using an [daft.io.AzureConfig][daft.io.AzureConfig] config object.

[daft.set_planning_config][daft.context.set_planning_config] is a convenient way to set your [daft.io.IOConfig][daft.io.IOConfig] as the default config to use on any subsequent Daft method calls.

=== "🐍 Python"

```python
from daft.io import IOConfig, AzureConfig

# Supply actual values for the storage_account and access key here
io_config = IOConfig(azure=AzureConfig(storage_account="***", access_key="***"))

# Globally set the default IOConfig for any subsequent I/O calls
daft.set_planning_config(default_io_config=io_config)

# Perform some I/O operation
df = daft.read_parquet("az://my_container/my_path/**/*")
```

Alternatively, Daft supports overriding the default IOConfig per-operation by passing it into the io_config= keyword argument. This is extremely flexible as you can pass a different [daft.io.AzureConfig][daft.io.AzureConfig] per function call if you wish!

=== "🐍 Python"

```python
# Perform some I/O operation but override the IOConfig
df2 = daft.read_csv("az://my_container/my_other_path/**/*", io_config=io_config)
```

Connect to Microsoft Fabric/OneLake Copy Section

If you are connecting to storage in OneLake or another Microsoft Fabric service, set the use_fabric_endpoint parameter to True in the [daft.io.AzureConfig][daft.io.AzureConfig] object.

=== "🐍 Python"

```python
from daft.io import IOConfig, AzureConfig

io_config = IOConfig(
    azure=AzureConfig(
        storage_account="onelake",
        use_fabric_endpoint=True,

        # Set credentials as needed
    )
)

df = daft.read_deltalake('abfss://[WORKSPACE]@onelake.dfs.fabric.microsoft.com/[LAKEHOUSE].Lakehouse/Tables/[TABLE]', io_config=io_config)
```

Connectors Bigtable (`docs/connectors/bigtable.md`) Copy Section

Writing to Google Cloud Bigtable

!!! warning "Experimental"

This connector is experimental and the API may change in future releases.

Google Cloud Bigtable is a fully managed, scalable NoSQL database service. Daft can write DataFrames to Bigtable tables using [df.write_bigtable()][daft.dataframe.DataFrame.write_bigtable].

Installing Dependencies Copy Chapter

Bigtable support requires the google-cloud-bigtable package:

pip install google-cloud-bigtable

Basic Usage Copy Chapter

import daft

# Create a DataFrame
df = daft.from_pydict({
    "user_id": ["user_001", "user_002", "user_003"],
    "name": ["Alice", "Bob", "Charlie"],
    "age": [30, 25, 35],
    "email": ["[email protected]", "[email protected]", "[email protected]"],
})

# Write to Bigtable
result = df.write_bigtable(
    project_id="my-gcp-project",
    instance_id="my-bigtable-instance",
    table_id="users",
    row_key_column="user_id",
    column_family_mappings={
        "name": "profile",
        "age": "profile",
        "email": "contact",
    },
)

Key Concepts Copy Chapter

Row Keys Copy Section

Every Bigtable row requires a unique row key. Use row_key_column to specify which DataFrame column should be used as the row key:

df.write_bigtable(
    ...,
    row_key_column="user_id",  # This column becomes the row key
)

Column Families Copy Section

Bigtable organizes columns into column families. Use column_family_mappings to specify which family each column belongs to:

df.write_bigtable(
    ...,
    column_family_mappings={
        "name": "user_data",      # 'name' column goes to 'user_data' family
        "age": "user_data",       # 'age' column goes to 'user_data' family
        "email": "contact_info",  # 'email' column goes to 'contact_info' family
    },
)

The column families must already exist in the Bigtable table.

Parameters Copy Chapter

Data Type Handling Copy Chapter

Bigtable cells only accept data that can be converted to bytes. By default, Daft automatically serializes incompatible types to JSON:

df = daft.from_pydict({
    "id": ["row1"],
    "data": [{"nested": "object"}],  # Complex type
})

# Complex types are automatically serialized to JSON
df.write_bigtable(
    ...,
    serialize_incompatible_types=True,  # Default behavior
)

To disable automatic serialization (will raise an error for incompatible types):

df.write_bigtable(
    ...,
    serialize_incompatible_types=False,
)

Advanced Configuration Copy Chapter

Client Options Copy Section

Pass additional options to the Bigtable client:

result = df.write_bigtable(
    ...,
    client_kwargs={
        "admin": True,
        "channel": custom_channel,
    },
)

Write Options Copy Section

Configure the MutationsBatcher for write operations:

result = df.write_bigtable(
    ...,
    write_kwargs={
        "flush_count": 1000,
        "max_row_bytes": 5 * 1024 * 1024,  # 5MB
    },
)

Use Cases Copy Chapter

IoT Data Storage Copy Section

import daft
from daft import col

# Read sensor data
df = daft.read_parquet("s3://bucket/sensors/*.parquet")

# Prepare for Bigtable (create composite row key)
df = df.with_column(
    "row_key",
    col("device_id").concat("#").concat(col("timestamp").cast(str)),
)

# Write to Bigtable
df.write_bigtable(
    project_id="iot-project",
    instance_id="sensor-data",
    table_id="readings",
    row_key_column="row_key",
    column_family_mappings={
        "temperature": "metrics",
        "humidity": "metrics",
        "device_id": "metadata",
        "timestamp": "metadata",
    },
)

User Profile Storage Copy Section

import daft

df = daft.from_pydict({
    "user_id": ["u001", "u002"],
    "preferences": [{"theme": "dark"}, {"theme": "light"}],
    "last_login": ["2024-01-15", "2024-01-16"],
})

df.write_bigtable(
    project_id="my-project",
    instance_id="user-store",
    table_id="profiles",
    row_key_column="user_id",
    column_family_mappings={
        "preferences": "settings",
        "last_login": "activity",
    },
)

Notes Copy Chapter

• The Bigtable table and column families must exist before writing
• Row keys should be designed carefully for efficient access patterns
• Consider Bigtable's row key design best practices for optimal performance

Connectors Blob (`docs/connectors/blob.md`) Copy Section

Reading Files as Blobs

Daft can read files as raw bytes into DataFrames using [daft.read_blob()][daft.io.read_blob], similar to DuckDB's read_blob function. This is useful for loading non-tabular files such as images, audio, PDFs, or any other binary data into a DataFrame, with one row per file.

Basic Usage Copy Chapter

Read files matching a path or glob pattern, where each file becomes a row:

=== "Local Files"

```python
import daft

df = daft.read_blob("/path/to/files/*.jpeg")
df.show()
```

=== "Remote Files (S3)"

```python
import daft
from daft.io import S3Config, IOConfig

io_config = IOConfig(s3=S3Config(region_name="us-west-2", anonymous=True))
df = daft.read_blob("s3://my-bucket/images/*.jpeg", io_config=io_config)
df.show()
```

=== "Remote Files (GCS)"

```python
import daft
from daft.io import GCSConfig, IOConfig

io_config = IOConfig(gcs=GCSConfig(anonymous=True))
df = daft.read_blob("gs://my-bucket/images/*.jpeg", io_config=io_config)
df.show()
```

Output Schema Copy Chapter

The read_blob function returns a DataFrame with the following columns:

Glob Patterns Copy Chapter

read_blob supports the same wildcards as other Daft readers:

# All .png files in a directory
df = daft.read_blob("/data/*.png")

# Recursive search
df = daft.read_blob("/data/**/*.png")

# Multiple paths
df = daft.read_blob(["/data/a.bin", "/data/b.bin"])

Error Handling Copy Chapter

By default, a download error for any file raises immediately. Set on_error="null" to log the error and fall back to a null content value instead:

df = daft.read_blob("/data/*.bin", on_error="null")

Common Use Cases Copy Chapter

Decoding Images Copy Section

Read image files and decode them for downstream processing:

import daft
from daft.functions import decode_image

df = daft.read_blob("s3://my-bucket/images/*.jpeg", io_config=io_config)
df = df.with_column("image", decode_image(df["content"]))

Detecting File Types Copy Section

Inspect magic bytes to determine the MIME type of each file:

import daft
from daft.functions import guess_mime_type

df = daft.read_blob("/data/**/*")
df = df.with_column("mime", guess_mime_type(df["content"]))

Relationship to Other APIs Copy Chapter

read_blob is a convenience API composed of [daft.from_glob_path()][daft.from_glob_path] (which lists files with their path and size) and the [download][daft.functions.download] expression (which fetches file contents as bytes). Use those building blocks directly if you need more control, such as downloading only a filtered subset of files.

Connectors Clickhouse (`docs/connectors/clickhouse.md`) Copy Section

Writing to ClickHouse

ClickHouse is a fast, open-source columnar database for real-time analytics. Daft can write DataFrames to ClickHouse tables using [df.write_clickhouse()][daft.dataframe.DataFrame.write_clickhouse].

Installing Dependencies Copy Chapter

ClickHouse support requires the clickhouse-connect package:

pip install clickhouse-connect

Basic Usage Copy Chapter

import daft

# Create a DataFrame
df = daft.from_pydict({
    "id": [1, 2, 3, 4],
    "name": ["Alice", "Bob", "Charlie", "Diana"],
    "value": [100.5, 200.3, 150.7, 300.2],
})

# Write to ClickHouse
result = df.write_clickhouse(
    table="my_table",
    host="localhost",
    port=8123,
    user="default",
    password="",
)
result.show()

Output Schema Copy Chapter

The write operation returns a DataFrame with write statistics:

Connection Parameters Copy Chapter

Advanced Configuration Copy Chapter

Client Options Copy Section

Pass additional options to the ClickHouse client:

result = df.write_clickhouse(
    table="my_table",
    host="localhost",
    client_kwargs={
        "secure": True,
        "verify": True,
        "connect_timeout": 30,
    },
)

Write Options Copy Section

Pass additional options to the write operation:

result = df.write_clickhouse(
    table="my_table",
    host="localhost",
    write_kwargs={
        "settings": {
            "async_insert": 1,
            "wait_for_async_insert": 1,
        },
    },
)

Use Cases Copy Chapter

Analytics Pipeline Copy Section

import daft
from daft import col

# Process data
df = daft.read_parquet("s3://bucket/events/*.parquet")
aggregated = df.groupby("event_type").agg(
    col("value").sum().alias("total_value"),
    col("user_id").count().alias("event_count"),
)

# Write aggregated results to ClickHouse for dashboards
aggregated.write_clickhouse(
    table="event_aggregates",
    host="clickhouse.example.com",
    database="analytics",
    user="writer",
    password="secret",
)

Real-Time Data Ingestion Copy Section

import daft

# Read streaming batch
df = daft.read_json("/data/batch/*.json")

# Transform and load to ClickHouse
df.write_clickhouse(
    table="events",
    host="localhost",
    write_kwargs={
        "settings": {"async_insert": 1},
    },
)

Notes Copy Chapter

• The target table must exist in ClickHouse before writing
• Column names and types in the DataFrame should match the table schema
• ClickHouse will perform type coercion where possible

Connectors Cos (`docs/connectors/cos.md`) Copy Section

Reading from and Writing to Tencent Cloud COS

Daft supports reading and writing data to Tencent Cloud COS (Cloud Object Storage), and understands natively the URL protocols cos:// and cosn:// (Hadoop CosN compatible) as referring to data that resides in COS.

Authorization/Authentication Copy Chapter

In Tencent Cloud COS, data is stored under the hierarchy of:

1. Bucket: The container for data storage, which is the top-level namespace for data storage in COS.
2. Object Key: The unique identifier for a piece of data within a bucket.

URLs to data in COS come in the form: cos://{BUCKET}/{OBJECT_KEY}.

!!! note "Hadoop CosN Compatibility"

Daft also supports the `cosn://` URL scheme for compatibility with Hadoop CosN. Both `cos://` and `cosn://` are treated identically.

Rely on Environment Copy Section

You can configure Daft to automatically discover credentials from environment variables. Daft supports the following environment variable prefixes:

Please be aware that when doing so in a distributed environment such as Ray, Daft will pick these credentials up from worker machines and thus each worker machine needs to be appropriately provisioned.

If instead you wish to have Daft use credentials from the "driver", you may wish to manually specify your credentials.

Manually specify credentials Copy Section

You may also choose to pass these values into your Daft I/O function calls using a [daft.io.CosConfig][daft.io.CosConfig] config object.

[daft.set_planning_config][daft.context.set_planning_config] is a convenient way to set your [daft.io.IOConfig][daft.io.IOConfig] as the default config to use on any subsequent Daft method calls.

=== "Using Permanent Keys"

```python
from daft.io import IOConfig, CosConfig

io_config = IOConfig(
    cos=CosConfig(
        region="ap-guangzhou",
        secret_id="your-secret-id",
        secret_key="your-secret-key",
    )
)

# Globally set the default IOConfig for any subsequent I/O calls
daft.set_planning_config(default_io_config=io_config)

# Perform some I/O operation
df = daft.read_parquet("cos://my-bucket/my_path/**/*")
```

=== "Using STS Temporary Credentials"

```python
from daft.io import IOConfig, CosConfig

io_config = IOConfig(
    cos=CosConfig(
        region="ap-guangzhou",
        secret_id="your-tmp-secret-id",
        secret_key="your-tmp-secret-key",
        security_token="your-security-token",
    )
)

# Perform some I/O operation
df = daft.read_parquet("cos://my-bucket/my_path/**/*", io_config=io_config)
```

=== "Using Custom Endpoint"

```python
from daft.io import IOConfig, CosConfig

# Use a custom endpoint (region will be auto-derived from the endpoint)
io_config = IOConfig(
    cos=CosConfig(
        endpoint="https://cos.ap-beijing.myqcloud.com",
        secret_id="your-secret-id",
        secret_key="your-secret-key",
    )
)

df = daft.read_parquet("cos://my-bucket/data/**/*", io_config=io_config)
```

=== "Anonymous Access"

```python
from daft.io import IOConfig, CosConfig

# Access public buckets without credentials
io_config = IOConfig(cos=CosConfig(anonymous=True))

# Read from a public bucket
df = daft.read_parquet("cos://public-bucket/data/**/*", io_config=io_config)
```

=== "Using CosN URL Scheme"

```python
from daft.io import IOConfig, CosConfig

io_config = IOConfig(
    cos=CosConfig(
        region="ap-guangzhou",
        secret_id="your-secret-id",
        secret_key="your-secret-key",
    )
)

# Use cosn:// scheme (Hadoop CosN compatible)
df = daft.read_parquet("cosn://my-bucket/my_path/**/*", io_config=io_config)
```

Alternatively, Daft supports overriding the default IOConfig per-operation by passing it into the io_config= keyword argument. This is extremely flexible as you can pass a different [daft.io.CosConfig][daft.io.CosConfig] per function call if you wish!

=== "Per-Operation Config"

```python
from daft.io import IOConfig, CosConfig

io_config = IOConfig(
    cos=CosConfig(
        region="ap-guangzhou",
        secret_id="your-secret-id",
        secret_key="your-secret-key",
    )
)

# Perform some I/O operation but override the IOConfig
df2 = daft.read_csv("cos://my-bucket/my_other_path/**/*", io_config=io_config)
```

Writing Data Copy Chapter

Daft supports writing data to COS using the same CosConfig:

import daft
from daft.io import IOConfig, CosConfig

io_config = IOConfig(
    cos=CosConfig(
        region="ap-guangzhou",
        secret_id="your-secret-id",
        secret_key="your-secret-key",
    )
)

df = daft.from_pydict({"a": [1, 2, 3], "b": ["x", "y", "z"]})

# Write to COS as Parquet
df.write_parquet("cos://my-bucket/output/", io_config=io_config)

Configuration Options Copy Chapter

The [daft.io.CosConfig][daft.io.CosConfig] object supports the following options:

!!! tip "Region and Endpoint Auto-Derivation"

You only need to specify either `region` or `endpoint` — Daft will automatically derive the other:

- If only `region` is provided, the endpoint is derived as `https://cos.{region}.myqcloud.com`
- If only `endpoint` is provided, the region is extracted from the endpoint URL
- If both are provided, both values are used as-is

Supported Operations Copy Chapter

Daft supports the following operations with COS:

• Read: read_parquet, read_csv, read_json, and other file readers
• Write: write_parquet, write_csv, write_json (including multipart uploads)
• List: Listing objects with glob pattern matching
• Delete: Deleting objects

Connectors Custom Catalogs (`docs/connectors/custom-catalogs.md`) Copy Section

Custom Catalogs

Daft ships with built-in catalog integrations for Iceberg, Unity, AWS Glue, and others. But if your tables live behind a proprietary metadata service, an internal API, or a convention like "every subdirectory in S3 is a Delta table," you can implement your own catalog and plug it into Daft. This also applies if you use a public catalog system that Daft doesn't have a built-in integration for yet - you can bridge the gap yourself rather than waiting for native support!

The Catalog and Table abstract base classes in daft.catalog are public interfaces designed for this. Subclass them to teach Daft how to discover and read your tables. Once registered, your catalog works everywhere - the Python API, SQL queries, and Sessions all resolve tables through it. Tables are resolved lazily, so you can expose thousands of tables without paying the cost of eagerly constructing DataFrames for all of them upfront.

Overview Copy Chapter

Daft's catalog system has two core interfaces:

• Catalog - discovers and manages tables and namespaces
• Table - reads and writes data for a single table

Both are abstract base classes. You implement the private _-prefixed abstract methods, and the base classes provide the public API (argument normalization, if_not_exists variants, etc.) on top.

Once implemented, custom catalogs work with Sessions and SQL just like the built-in ones:

import daft
from daft.session import Session

sess = Session()
sess.attach(my_catalog)
sess.sql("SELECT * FROM my_namespace.my_table").show()

Catalog Interface Copy Chapter

A Catalog subclass implements these abstract methods:

The base class provides the public methods (get_table, list_tables, create_table_if_not_exists, read_table, write_table, etc.) on top of these.

Table Interface Copy Chapter

A Table subclass implements these abstract methods:

The base class provides write(df, mode), select(*columns), and show(n) on top of these.

Walkthrough: Iceberg Catalog Implementation Copy Chapter

Daft's Iceberg Catalog integration is a complete built-in implementation that covers every abstract method. The snippets below are trimmed to highlight the pattern - see the full source for details.

IcebergCatalog Copy Section

The catalog wraps a PyIceberg Catalog client. Each abstract method converts between Daft's Identifier type and PyIceberg's tuple-based identifiers, then delegates to the inner client:

/* Detailed source-code truncated for AI context efficiency. */

The key pattern: convert Identifier to your backend's native format (here, a tuple via tuple(identifier)), call the backend, and translate exceptions to NotFoundError.

IcebergTable Copy Section

The table wraps a PyIceberg Table object. read() delegates to daft.read_iceberg and write methods delegate to DataFrame.write_iceberg:

from daft.catalog import Table
from daft.dataframe import DataFrame
from daft.logical.schema import Schema

class IcebergTable(Table):
    _inner: InnerTable  # pyiceberg.table.Table

    @property
    def name(self) -> str:
        return self._inner.name()[-1]

    def schema(self) -> Schema:
        # Iceberg derives schema from the lazy read plan here, but if your
        # backend exposes schema metadata directly, prefer that instead.
        return self.read().schema()

    def read(self, **options) -> DataFrame:
        return daft.read_iceberg(self._inner, **options)

    def append(self, df: DataFrame, **options) -> None:
        df.write_iceberg(self._inner, mode="append")

    def overwrite(self, df: DataFrame, **options) -> None:
        df.write_iceberg(self._inner, mode="overwrite")

Using a Custom Catalog Copy Chapter

Custom catalogs integrate with the full Daft API - Sessions, SQL, and the Python API:

import daft
from daft.session import Session

# Create your catalog
catalog = MyCatalog(...)

# Use it directly
catalog.list_tables()
catalog.read_table("my_namespace.my_table")

# Or attach it to a session for SQL support
sess = Session()
sess.attach(catalog)
sess.set_namespace("my_namespace")

# SQL queries resolve tables through your catalog
sess.sql("SELECT * FROM my_table").show()

Reference Copy Chapter

For complete API documentation, see the Catalogs & Tables API reference.

Other built-in implementations for reference:

• Unity Catalog - read-only, simplest implementation
• AWS Glue Catalog - AWS Glue metadata service
• PostgreSQL Catalog - full read/write with PostgreSQL

Connectors Custom (`docs/connectors/custom.md`) Copy Section

Reading from and Writing to Custom Connectors

This guide shows you how to create custom connectors to read from or write to data sources that aren't currently implemented in Daft, which includes proprietary data sources that can't be implemented in open source.

Overview Copy Chapter

Daft provides two main interfaces for custom connectors:

• Reading from a custom DataSource
• Writing to a custom DataSink

You can also take a look at actual code references on how we implemented:

• Reads from Video Frames
• Writes to Hugging Face 🤗
• More examples below

Reading from a Custom Data Source Copy Chapter

Step 1: Implement the `DataSource` and `DataSourceTask` Interfaces Copy Section

Create a class that inherits from DataSource, a class that inherits from DataSourceTask, and implement the required methods. Here's a simple example doing this with a custom local file reader that reads each line from a file as a single String row.

=== "🐍 Python"

/* Detailed source-code truncated for AI context efficiency. */

Step 2: Use Your Custom Data Source to Create a Daft DataFrame Copy Section

=== "🐍 Python"

import daft

# Create a sample text file and read from it.
sample_file = "sample_text.txt"
with open(sample_file, "w") as f:
    f.write("Alice was beginning to get very tired of sitting by her sister on the bank.\n")
    f.write("So she was considering in her own mind (as well as she could, for the hot day made her feel very sleepy and stupid),\n")
    f.write("when suddenly a White Rabbit with pink eyes ran close by her.\n")
    f.write("There was nothing so very remarkable in that;\n")
    f.write("nor did Alice think it so very much out of the way to hear the Rabbit say to itself, 'Oh dear! Oh dear! I shall be late!'\n")

data_source = TextFileDataSource([sample_file])

(
    data_source
    .read()
    .show()
)

Output:

╭────────────────────────────────╮
│ line                           │
│ ---                            │
│ Utf8                           │
╞════════════════════════════════╡
│ Alice was beginning to get ve… │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ So she was considering in her… │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ when suddenly a White Rabbit … │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ There was nothing so very rem… │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤
│ nor did Alice think it so ver… │
╰────────────────────────────────╯

(Showing first 5 of 5 rows)

Optional: Declaring Clustering to Skip Shuffles Copy Section

Clustering is an execution-time property: it describes how rows are distributed across the
in-memory partitions a query runs over. It is distinct from storage partitioning (the on-disk
layout declared via get_partition_fields(), e.g. Hive/Iceberg directories) — a source can be laid
out one way on disk yet emit partitions clustered another way.

Two clustering hints are available:

Hash clustering

If your source already emits data that is hash-partitioned by some keys — for example, each
DataSourceTask corresponds to exactly one
(producer, hour) group — you can tell Daft by overriding get_clustering_keys(). Daft then
skips the shuffle it would otherwise insert before a downstream groupby, Window.partition_by,
or distinct whose keys are covered by (equal to, or a superset of) the declared clustering.

from daft import col
from daft.io.clustering import ClusteringKeys
from daft.io.source import DataSource


class ClusteredSource(DataSource):
    # ... name / schema / get_tasks as above ...

    def get_clustering_keys(self) -> ClusteringKeys | None:
        # Each task emits exactly one (a, b) group, so the output is hash-partitioned by (a, b).
        return ClusteringKeys.hash("a", "b")

Keys may be column names or arbitrary Expressions. An expression-valued
key follows a projection that materializes it as a derived column, so the clustering is preserved
even after a with_column:

def hour_bucket(ts: "daft.Expression") -> "daft.Expression":
    # bucket a unix-epoch timestamp (in seconds) into hourly buckets
    return ts // 3600


class EventSource(DataSource):
    def get_clustering_keys(self) -> ClusteringKeys | None:
        return ClusteringKeys.hash(col("producer"), hour_bucket(col("ts")))


df = (
    EventSource().read()
    .with_column("hour", hour_bucket(col("ts")))  # materialize the expression key as a column
    .groupby("producer", "hour")                   # covered by the declared clustering => no shuffle
    .sum("value")
)

Range clustering

If your source already emits data where each task covers a non-overlapping range of values for
those columns, you can tell Daft by returning ClusteringKeys.range(). Pass descending=True
if partitions are ordered from highest to lowest values. If any key values are null, declare
where those rows live with nulls_first (default: nulls in the last partition when ascending,
the first when descending, matching sort semantics).

class TimeSeriesSource(DataSource):
    # ... name / schema / get_tasks as above ...

    def get_clustering_keys(self) -> ClusteringKeys | None:
        # Each task holds a non-overlapping ts range (ascending partition order).
        return ClusteringKeys.range("ts")

    # Or, if partitions are ordered high-to-low:
    # return ClusteringKeys.range("ts", descending=True)

!!! note "Shuffle elision applies to the distributed runner"

Daft only inserts these shuffles when running distributed (e.g. on Ray); the single-node
runner already executes the operators locally. The declaration is therefore a no-op for local
execution and only changes plans for distributed runs.

!!! warning "Clustering must hold for every task"

Daft trusts the declaration. For hash clustering, every row with the same hash of the declared
keys must be produced within a single task. For range clustering, each task must cover a
genuinely non-overlapping range of values for the declared columns. Incorrectly overriding
`get_clustering_keys()` will likely lead to incorrect results.

Writing to a Custom Data Sink Copy Chapter

Step 1: Implement the `DataSink` Interface Copy Section

Create a class that inherits from DataSink and implements the required methods. Here's a simple example doing this with a custom local file writer.

=== "🐍 Python"

/* Detailed source-code truncated for AI context efficiency. */

Step 2: Use Your Custom Data Sink Copy Section

=== "🐍 Python"

import daft
from custom_datasink_example import LocalFileDataSink


data = {
    "id": list(range(1, 1001)),
    "name": [f"User_{i}" for i in range(1, 1001)],
}

local_file_data_sink = LocalFileDataSink(
    output_dir="./output_folder",
    filename_prefix="users",
    max_rows_per_file=10
)

(
    daft.from_pydict(data)
    .write_sink(local_file_data_sink)
    .show()
)

Sample File Output:

{'id': 111, 'name': 'User_111'}
{'id': 112, 'name': 'User_112'}
{'id': 113, 'name': 'User_113'}
{'id': 114, 'name': 'User_114'}
{'id': 115, 'name': 'User_115'}
{'id': 116, 'name': 'User_116'}
{'id': 117, 'name': 'User_117'}
{'id': 118, 'name': 'User_118'}
{'id': 119, 'name': 'User_119'}
{'id': 120, 'name': 'User_120'}

More Examples Copy Chapter

For further reference, feel free to check out some of our notable data connectors that are implemented using the DataSource and DataSink interfaces:

• Reads from Video Frames
• Reads from LanceDB
• Writes to Hugging Face 🤗
• Writes to ClickHouse
• Writes to Turbopuffer
• Writes to LanceDB

Connectors Delta Lake (`docs/connectors/delta_lake.md`) Copy Section

Reading from and Writing to Delta Lake

Delta Lake is an open-source storage framework for data analytics on data lakes. It provides ACID transactions, scalable metadata handling, and a unification of streaming and batch data processing, all on top of Parquet files in cloud storage.

Daft currently supports:

1. Parallel + Distributed Reads: Daft parallelizes Delta Lake table reads over all cores of your machine, if using the default multithreading runner, or all cores + machines of your Ray cluster, if using the distributed Ray runner.

2. Skipping Filtered Data: Daft ensures that only data that matches your [df.where()][daft.DataFrame.where] filter will be read, often skipping entire files/partitions.

3. Multi-cloud Support: Daft supports reading Delta Lake tables from AWS S3, Azure Blob Store, and GCS, as well as local files.

A detailed Delta Lake roadmap for Daft can be found on our GitHub issues. For the overall Daft development plan, see Daft Roadmap.

Installing Daft with Delta Lake Support Copy Chapter

Daft internally uses the deltalake Python package to fetch metadata about the Delta Lake table, such as paths to the underlying Parquet files and table statistics. The deltalake package therefore must be installed to read Delta Lake tables with Daft, either manually or with the below daft[deltalake] extras install of Daft.

pip install -U "daft[deltalake]"

Reading a Table Copy Chapter

A Delta Lake table can be read by providing [daft.read_deltalake][daft.read_deltalake] with the URI for your table.

The below example uses the deltalake Python package to create a local Delta Lake table for Daft to read, but Daft can also read Delta Lake tables from all of the major cloud stores.

=== "🐍 Python"

```python
# Create a local Delta Lake table.
from deltalake import write_deltalake
import pandas as pd

df = pd.DataFrame({
    "group": [1, 1, 2, 2, 3, 3, 4, 4],
    "num": list(range(8)),
    "letter": ["a", "b", "c", "d", "e", "f", "g", "h"],
})

# This will write out separate partitions for group=1, group=2, group=3, group=4.
write_deltalake("some-table", df, partition_by="group")
```

After writing this local example table, we can easily read it into Daft.

=== "🐍 Python"

```python
# Read a Delta Lake table
import daft

df = daft.read_deltalake("some-table")
```

Data Skipping Optimizations Copy Chapter

Subsequent filters on the partition column group will efficiently skip data that doesn't match the predicate. In the below example, the group != 2 partitions (files) will be pruned, i.e. they will never be read into memory.

=== "🐍 Python"

```python
# Filter on partition columns will result in efficient partition pruning; non-matching partitions will be skipped.
df2 = df.where(df["group"] == 2)
df2.show()
```

Filters on non-partition columns will still benefit from automatic file pruning via file-level statistics. In the below example, the group=2 partition (file) will have 2 <= df["num"] <= 3 lower/upper bounds for the num column, and since the filter predicate is df["num"] < 2, Daft will prune the file from the read. Similar is true for group=3 and group=4 partitions, with none of the data from those files being read into memory.

=== "🐍 Python"

```python
# Filter on non-partition column, relying on file-level column stats to efficiently prune unnecessary file reads.
df3 = df.where(df["num"] < 2)
df3.show()
```

Write to Delta Lake Copy Chapter

You can use [df.write_deltalake()][daft.DataFrame.write_deltalake] to write a DataFrame to a Delta table:

=== "🐍 Python"

```python
df.write_deltalake("tmp/daft-recordbatch", mode="overwrite")
```

Daft supports multiple write modes. See the API docs for [df.write_deltalake()][daft.DataFrame.write_deltalake] for more details.

When writing Delta Lake tables to S3, Daft relies on the native conditional write support in deltalake versions >=0.23.0. DynamoDB locking is not required by default for AWS S3 writes. To explicitly use a DynamoDB locking provider, pass dynamo_table_name="..." to [df.write_deltalake()][daft.DataFrame.write_deltalake]. For deltalake versions less than 0.23.0, if dynamo_table_name is not provided and allow_unsafe_rename is False, ValueError exception will be raised.

Checkpointing Copy Chapter

Daft supports idempotent writes to Delta Lake via the checkpoint= parameter on [df.write_deltalake()][daft.DataFrame.write_deltalake]. Retries of the same logical commit — after a crash, a transient catalog error, or a deliberate re-invocation — produce the same Delta state without duplicate commits. See the Checkpointing user guide for concepts; the sections below cover Delta-specific behavior.

Example Copy Section

The pattern: one CheckpointStore paired into both the source (via CheckpointConfig) and the sink (via IdempotentCommit). The source records which inputs were processed; the sink stamps the resulting Delta commit with the idempotence key.

=== "🐍 Python"

```python
import daft

# One store, paired into both the source and the sink.
store = daft.CheckpointStore("s3://my-bucket/ckpt/")

df = daft.read_parquet(
    "s3://input/",
    checkpoint=daft.CheckpointConfig(store, on="file_id"),
)

# Any map-only operations work here: filter, project, UDF, explode, ...
df = df.where(df["status"] == "active")

written = df.write_deltalake(
    "s3://my-bucket/my-table/",
    checkpoint=daft.IdempotentCommit(store, idempotence_key="job-2026-05-21-001"),
)
```

A fresh run produces one new Delta commit tagged with daft.idempotence-key=job-2026-05-21-001. Retries with the same idempotence_key recognize the prior commit and exit cleanly — no duplicate commit, no reprocessing of inputs already handled.

Inspecting the Marker Copy Section

Every idempotent commit tags its Delta commit info with daft.idempotence-key. To verify which logical commit produced the current state of a table:

=== "🐍 Python"

```python
# Inspect via the deltalake library.
from deltalake import DeltaTable
table = DeltaTable("s3://my-bucket/my-table/")
print(table.history()[0].get("daft.idempotence-key"))
# → "job-2026-05-21-001"
```

For older commits, walk table.history() and check each entry.

Constraints Copy Section

These are constraints specific to the Delta Lake connector. Daft-wide constraints (Ray runner, map-only pipelines, single-writer concurrency, etc.) are in the user guide's Limitations section.

• mode='append' only. Other modes raise NotImplementedError when checkpoint= is set.
• Reserved daft.idempotence-* property prefix. Keys in custom_metadata that start with this prefix raise ValueError — Daft uses this namespace internally.

Idempotence-Key Contract Copy Section

The user picks the idempotence_key. Daft uses it as the marker that identifies a logical commit. The key must be stable across retries and unique across distinct logical commits — both halves matter:

• Same key, different inputs → silent no-op (data loss). Daft sees the existing marker and skips the commit. The new data is dropped without an error.
• Different key on a retry of the same logical commit → duplicate commit. Daft doesn't recognize the prior attempt and lands a second commit.

Recovery Copy Section

The user guide's Recovery section walks through the full chronology. The Delta-specific steps:

• Commit-from-store. When a run crashes after the pipeline finishes but before the Delta commit lands, Daft reads the staged file references from the store on rerun and commits them as a single new Delta transaction with daft.idempotence-key in custom_metadata.

• Marker recognition. When a run crashes after the Delta commit lands but before the store is marked done — or when the user deliberately re-runs with the same key — Daft walks table.history(), finds the marker, marks the store, and exits. No second commit. Returned DataFrame is empty.

Orphan files on crash. A worker that crashed mid-task may leave parquet files unreferenced by any Delta commit. Daft doesn't auto-clean these — use the deltalake vacuum operation.

Delta-Specific Notes Copy Section

• Parallel history walk. table.history() is already concurrent under the hood (deltalake-rs buffers up to num_cpus * 4 log reads), so marker recognition is fast even on tables with long commit histories.

• Fresh-table case handled. If the target table doesn't exist yet at the configured URI, the first idempotent write creates it.

• Transient retry on CommitFailedError. Daft retries up to twice on this exception (concurrent-writer conflicts, lock contention). Other exceptions — schema mismatches, metadata-construction errors, network errors — propagate immediately. Wrap the call in your own retry policy if you need broader coverage.

Type System Copy Chapter

Daft and Delta Lake have compatible type systems. Here are how types are converted across the two systems.

When reading from a Delta Lake table into Daft:

References:

• Python unitycatalog type name code reference
• Spark types documentation
• Databricks types documentation

Connectors Files (`docs/connectors/files.md`) Copy Section

Creating File References with from_files

Daft provides [daft.from_files()][daft.io.from_files] to create a DataFrame of lazy file references from glob patterns. Unlike other read functions that immediately load file contents, from_files creates File objects that can be read on demand.

Basic Usage Copy Chapter

=== "Local Files"

```python
import daft

df = daft.from_files("/path/to/files/*.jpeg")
df.show()
```

=== "Remote Files (S3)"

```python
import daft

df = daft.from_files("s3://my-bucket/images/*.png")
df.show()
```

=== "Remote Files (GCS)"

```python
import daft

df = daft.from_files("gs://my-bucket/images/*.png")
df.show()
```

Output Schema Copy Chapter

The from_files function returns a DataFrame with a single column:

Wildcard Patterns Copy Chapter

from_files supports standard glob patterns: