Skip to content

datafusion-dataframe.md

Latest commit

 

History

History
260 lines (186 loc) · 7.56 KB

datafusion-dataframe.md

File metadata and controls

260 lines (186 loc) · 7.56 KB
Raw

DataFusion DataFrame Platform

DataFusion is an Apache Arrow-native query engine with a Rust backend and Python bindings. BenchBox supports benchmarking DataFusion using its native DataFrame API through the datafusion-df platform.

Overview

Attribute Value
CLI Name datafusion-df
Family Expression
Execution Lazy
Best For Medium to large datasets with complex analytical queries
Min Version 40.0.0

Features

  • Arrow-native - Zero-copy data sharing with other Arrow tools
  • Lazy evaluation - Query optimization before execution
  • Predicate pushdown - Efficient Parquet filtering
  • Rust performance - High-performance query execution
  • SQL support - Hybrid DataFrame/SQL queries

Installation

uv add datafusion

pip install datafusion

python -c "import datafusion; print(f'DataFusion {datafusion.__version__}')"

Quick Start

benchbox run --platform datafusion-df --benchmark tpch --scale 0.01

benchbox run --platform datafusion-df --benchmark tpch --scale 1 \
  --platform-option target_partitions=8

The second command runs with custom parallelism.

Configuration Options

Option Default Description
target_partitions CPU count Number of partitions for parallelism
repartition_joins true Enable automatic repartitioning for joins
parquet_pushdown true Enable predicate/projection pushdown for Parquet
batch_size 8192 Batch size for query execution
memory_limit None Memory limit for fair spill pool (e.g., '8G', '16GB')
temp_dir System temp Temporary directory for disk spilling

target_partitions

Controls parallelism by setting the number of partitions for query execution:

benchbox run --platform datafusion-df --benchmark tpch --scale 1 \
  --platform-option target_partitions=4

benchbox run --platform datafusion-df --benchmark tpch --scale 10

The first command uses 4 partitions, which suits limited memory. The second uses all available CPUs (the default).

parquet_pushdown

DataFusion can push predicates and projections down to the Parquet reader:

benchbox run --platform datafusion-df --benchmark tpch --scale 1 \
  --platform-option parquet_pushdown=false

Disable pushdown for debugging.

Scale Factor Guidelines

DataFusion uses lazy evaluation and can handle larger datasets than eager frameworks:

Scale Factor Data Size Memory Required Use Case
0.01 ~10 MB ~200 MB Unit testing, CI/CD
0.1 ~100 MB ~500 MB Integration testing
1.0 ~1 GB ~2 GB Standard benchmarking
10.0 ~10 GB ~8 GB Performance testing
100.0 ~100 GB ~32 GB Large-scale testing

Note: Execution times vary based on query complexity, hardware, and configuration. Run benchmarks to establish baselines for your environment.

Performance Characteristics

Strengths

  • Query optimization - Lazy evaluation enables automatic optimization
  • Memory efficiency - Streaming execution reduces memory pressure
  • Parquet performance - Native Parquet support with pushdown
  • Arrow integration - Zero-copy interop with PyArrow ecosystem

Limitations

  • Single-node - No distributed execution (unlike Spark)
  • Python overhead - Some overhead in Python bindings
  • Smaller community - Less documentation than Pandas/Polars

Performance Tips

  1. Enable parallelism for large datasets:

    --platform-option target_partitions=16

  2. Use Parquet format for best performance:

    benchbox run --platform datafusion-df --benchmark tpch --scale 10

  3. Compare with Polars for single-node workloads:

    --platform polars-df

Query Implementation

DataFusion queries use expression-based operations:

This is TPC-H Q1, the Pricing Summary Report. It filters with the expression API, computes derived columns and then aggregates.

def q1_datafusion_impl(ctx: DataFrameContext) -> Any:
    lineitem = ctx.get_table("lineitem")

    cutoff = ctx.cast_date(ctx.lit("1998-09-02"))
    filtered = lineitem.filter(ctx.col("l_shipdate") <= cutoff)

    disc_price = ctx.col("l_extendedprice") * (ctx.lit(1) - ctx.col("l_discount"))
    charge = disc_price * (ctx.lit(1) + ctx.col("l_tax"))

    result = (
        filtered
        .aggregate(
            [ctx.col("l_returnflag"), ctx.col("l_linestatus")],
            [
                f.sum(ctx.col("l_quantity")).alias("sum_qty"),
                f.avg(ctx.col("l_quantity")).alias("avg_qty"),
                f.sum(disc_price).alias("sum_disc_price"),
                f.sum(charge).alias("sum_charge"),
            ]
        )
        .sort(ctx.col("l_returnflag").sort(ascending=True))
    )

    return result

Comparison: DataFusion vs Polars DataFrame

Aspect DataFusion (datafusion-df) Polars (polars-df)
Backend Rust (via Arrow) Rust
Execution Lazy Lazy
Memory Efficient Very efficient
Speed Sub-second queries Sub-100ms queries
SQL Support Full SQL engine Basic SQL
Best for SQL + DataFrame hybrid Pure DataFrame

Window Functions

DataFusion supports SQL-style window functions:

from benchbox.platforms.dataframe import DataFusionDataFrameAdapter

adapter = DataFusionDataFrameAdapter()

row_num = adapter.window_row_number(
    order_by=[("sale_date", True)],
    partition_by=["category"]
)

running_total = adapter.window_sum(
    column="amount",
    partition_by=["category"],
    order_by=[("sale_date", True)]
)

Troubleshooting

Memory Issues

DataFusion error: Memory exhausted

Solutions:

  1. Set a memory limit with spilling: --platform-option memory_limit=8G
  2. Reduce partitions: --platform-option target_partitions=4
  3. Reduce batch size: --platform-option batch_size=4096
  4. Set a temp directory for spilling: --platform-option temp_dir=/tmp/datafusion
  5. Reduce scale factor: --scale 1

Slow Parquet Reads

If Parquet file reads are slow:

--platform-option parquet_pushdown=true

Type Errors

DataFusion is strict about types. Ensure date columns are properly cast:

date_col = adapter.cast_date(adapter.col("date_string"))

Python API

from benchbox.platforms.dataframe import DataFusionDataFrameAdapter

adapter = DataFusionDataFrameAdapter(
    working_dir="./benchmark_data",
    target_partitions=8,
    repartition_joins=True,
    parquet_pushdown=True,
    batch_size=8192,
    memory_limit="8G",
    temp_dir="/tmp/datafusion"
)

ctx = adapter.create_context()
adapter.load_tables(ctx, data_dir="./tpch_data")

df = adapter.sql("SELECT * FROM lineitem WHERE l_quantity > 10")
result = adapter.collect(df)

from benchbox.core.tpch.dataframe_queries import TPCH_DATAFRAME_QUERIES
query = TPCH_DATAFRAME_QUERIES.get_query("Q1")
result = adapter.execute_query(ctx, query)
print(result)

memory_limit enables memory management with spilling, and temp_dir is the directory for spilled data.

Related Documentation