Data Engineering with Python: Polars and DuckDB

Data Engineering with Python: Polars and DuckDB

Pandas is the standard tool for data analysis in Python, but when datasets grow to hundreds of millions of rows or you need complex SQL queries with minimal latency, Polars and DuckDB are the modern alternatives you need to know. Both are written in Rust/C++ and leverage columnar processing to be 10-100× faster than pandas on aggregation and filtering operations.

Why Polars and DuckDB

pip install polars duckdb pyarrow

Polars: Fast DataFrames with Lazy Evaluation

Basic eager operations

import polars as pl

# Read CSV (much faster than pandas)
df = pl.read_csv("sales.csv")

# Selection and filtering
result = df.filter(
    (pl.col("amount") > 1000) & (pl.col("country") == "US")
).select([
    "date", "customer", "amount", "product"
]).sort("amount", descending=True)

print(result.head(10))
print(result.describe())

Lazy evaluation: Polars' superpower

# LazyFrame — executes nothing until .collect()
lf = pl.scan_csv("sales_100M.csv")  # 100M row file

pipeline = (
    lf
    .filter(pl.col("year") >= 2023)
    .group_by(["country", "product"])
    .agg([
        pl.col("amount").sum().alias("total"),
        pl.col("amount").mean().alias("avg"),
        pl.col("customer").n_unique().alias("unique_customers"),
        pl.col("amount").count().alias("transactions"),
    ])
    .sort("total", descending=True)
    .limit(50)
)

# View the optimized execution plan
print(pipeline.explain())

# Execute (this is where processing happens)
result = pipeline.collect()
print(result)

Expressions and window functions

df = pl.read_parquet("data.parquet")

df = df.with_columns([
    # Ratio over global mean
    (pl.col("sales") / pl.col("sales").mean()).alias("ratio_mean"),

    # Rank by group (window function)
    pl.col("sales")
      .rank(method="dense", descending=True)
      .over("region")
      .alias("region_rank"),

    # 7-day rolling average
    pl.col("sales")
      .rolling_mean(window_size=7, min_periods=1)
      .alias("rolling_7d"),

    # Day-over-day change
    pl.col("sales").diff(1).alias("sales_delta"),

    # Conditional classification
    pl.when(pl.col("sales") > 10000)
      .then(pl.lit("high"))
      .when(pl.col("sales") > 5000)
      .then(pl.lit("medium"))
      .otherwise(pl.lit("low"))
      .alias("segment"),
])

Joins and merges

customers = pl.read_csv("customers.csv")
orders    = pl.read_csv("orders.csv")
products  = pl.read_csv("products.csv")

result = (
    orders
    .join(customers, on="customer_id", how="left")
    .join(products, on="product_id", how="inner")
    .filter(pl.col("category") == "electronics")
    .group_by("country")
    .agg(pl.col("amount").sum().alias("electronics_total"))
    .sort("electronics_total", descending=True)
)

Efficient file formats

# Parquet — columnar compressed format (recommended for large data)
df.write_parquet("data.parquet", compression="zstd")
df2 = pl.read_parquet("data.parquet")

# Read multiple Parquet files (glob)
df_all = pl.read_parquet("data_partition_*.parquet")

# Arrow IPC (fastest for inter-process transfer)
df.write_ipc("data.arrow")
df3 = pl.read_ipc("data.arrow")

DuckDB: Embedded OLAP SQL in Python

DuckDB is an in-process analytical database — no server needed. It works directly on pandas/Polars DataFrames, Parquet files, CSV, JSON, and in-memory data.

Basic usage

import duckdb

# In-memory connection (default)
con = duckdb.connect()

# Or persisted to disk
con = duckdb.connect("analytics.duckdb")

# Direct query over file
result = con.execute("""
    SELECT
        country,
        SUM(amount) as total,
        COUNT(*) as orders,
        AVG(amount) as avg_order
    FROM 'sales.csv'
    WHERE year >= 2023
    GROUP BY country
    ORDER BY total DESC
    LIMIT 20
""").df()  # .df() → pandas, .pl() → Polars

print(result)

DuckDB + Polars (native integration)

import duckdb
import polars as pl

df_polars = pl.read_parquet("sales.parquet")

con = duckdb.connect()
result = con.execute("""
    SELECT
        product,
        region,
        SUM(sales) as total_sales,
        RANK() OVER (PARTITION BY region ORDER BY SUM(sales) DESC) as rank
    FROM df_polars
    GROUP BY product, region
    QUALIFY rank <= 3
""").pl()

print(result)

Querying multiple Parquet files

con = duckdb.connect()

result = con.execute("""
    SELECT
        YEAR(date) as year,
        MONTH(date) as month,
        SUM(amount) as revenue,
        COUNT(DISTINCT customer_id) as active_customers
    FROM read_parquet('data/year=*/month=*/*.parquet', hive_partitioning=true)
    WHERE year BETWEEN 2022 AND 2024
    GROUP BY 1, 2
    ORDER BY 1, 2
""").df()

Advanced window functions and CTEs

query = """
WITH cumulative_sales AS (
    SELECT
        date,
        region,
        amount,
        SUM(amount) OVER (
            PARTITION BY region
            ORDER BY date
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as cumulative,
        LAG(amount, 1, 0) OVER (PARTITION BY region ORDER BY date) as prev_day,
        AVG(amount) OVER (
            PARTITION BY region
            ORDER BY date
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) as rolling_7d_avg
    FROM sales
),
with_change AS (
    SELECT *,
        ROUND(100.0 * (amount - prev_day) / NULLIF(prev_day, 0), 2) as pct_change
    FROM cumulative_sales
)
SELECT * FROM with_change
WHERE date >= CURRENT_DATE - INTERVAL 30 DAY
ORDER BY date DESC, region
"""

df = con.execute(query).df()

Complete ETL pipeline with DuckDB

import duckdb
import polars as pl
from pathlib import Path
import time

def etl_pipeline(input_folder: str, output_file: str):
    """ETL Pipeline: raw CSVs → clean, aggregated Parquet."""
    con = duckdb.connect()
    start = time.perf_counter()

    # 1. Extract
    con.execute(f"""
        CREATE TABLE raw AS
        SELECT * FROM read_csv_auto('{input_folder}/*.csv',
            union_by_name=true,
            ignore_errors=true)
    """)

    n_raw = con.execute("SELECT COUNT(*) FROM raw").fetchone()[0]
    print(f"Rows extracted: {n_raw:,}")

    # 2. Transform
    con.execute("""
        CREATE TABLE clean AS
        SELECT
            CAST(date AS DATE)                        AS date,
            UPPER(TRIM(country))                      AS country,
            COALESCE(region, 'UNKNOWN')               AS region,
            CAST(REPLACE(amount, ',', '.') AS DOUBLE) AS amount,
            CAST(quantity AS INTEGER)                 AS quantity,
            amount * quantity                         AS total_amount,
            MD5(CONCAT(date, customer_id, product_id)) AS row_id
        FROM raw
        WHERE date IS NOT NULL
          AND amount > 0
          AND quantity > 0
        QUALIFY ROW_NUMBER() OVER (PARTITION BY row_id ORDER BY date) = 1
    """)

    n_clean = con.execute("SELECT COUNT(*) FROM clean").fetchone()[0]
    print(f"Clean rows: {n_clean:,} ({100*n_clean/n_raw:.1f}% retained)")

    # 3. Load
    con.execute(f"""
        COPY (
            SELECT * FROM clean
            ORDER BY country, date
        ) TO '{output_file}'
        (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
    """)

    elapsed = time.perf_counter() - start
    size_mb = Path(output_file).stat().st_size / 1_048_576
    print(f"Written: {size_mb:.1f} MB in {elapsed:.2f}s")

    return con.execute("SELECT * FROM clean LIMIT 5").df()

sample = etl_pipeline("data/raw", "data/clean.parquet")
print(sample)

Benchmark: pandas vs Polars vs DuckDB

import pandas as pd
import polars as pl
import duckdb
import time
import numpy as np

# Generate test dataset (10M rows)
np.random.seed(42)
n = 10_000_000

data = {
    "id": np.arange(n),
    "country": np.random.choice(["US", "GB", "DE", "FR", "ES"], n),
    "amount": np.random.exponential(500, n),
    "category": np.random.choice(["A", "B", "C", "D"], n),
}

def benchmark(name, fn):
    start = time.perf_counter()
    result = fn()
    elapsed = time.perf_counter() - start
    print(f"{name:20s}: {elapsed:.3f}s")
    return elapsed

df_pd = pd.DataFrame(data)
benchmark("pandas", lambda:
    df_pd.groupby(["country", "category"])["amount"]
         .agg(["sum", "mean", "count"])
         .reset_index()
)

df_pl = pl.DataFrame(data)
benchmark("polars eager", lambda:
    df_pl.group_by(["country", "category"])
         .agg([pl.col("amount").sum(), pl.col("amount").mean(), pl.col("amount").count()])
)

benchmark("duckdb", lambda:
    duckdb.execute("""
        SELECT country, category, SUM(amount), AVG(amount), COUNT(*)
        FROM df_pd GROUP BY country, category
    """).df()
)

# Typical results on a modern laptop:
# pandas          :  2.841s
# polars eager    :  0.187s  (15× faster)
# duckdb          :  0.142s  (20× faster)

When to Use Each Tool

Ecosystem Integration

# Polars → pandas (when a library only supports pandas)
df_pandas = df_polars.to_pandas()
df_polars = pl.from_pandas(df_pandas)

# Polars → Arrow
arrow_table = df_polars.to_arrow()
df_polars2  = pl.from_arrow(arrow_table)

# DuckDB → Parquet directly
con.execute("COPY table TO 'output.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")

# Read from S3 (with httpfs extension)
con.execute("INSTALL httpfs; LOAD httpfs;")
df = con.execute("""
    SELECT * FROM read_parquet('s3://my-bucket/data/*.parquet')
    LIMIT 1000
""").pl()

Conclusion

Polars and DuckDB are complementary tools that cover most modern data engineering needs without requiring a Spark cluster. Polars excels at chained transformations with lazy evaluation, while DuckDB shines at analytical ad-hoc SQL queries. Both integrate seamlessly with the Arrow/Parquet ecosystem, utilize all CPU cores, and consume far less memory than pandas for aggregation operations. For Python data pipelines in 2025, this combination is the recommended starting point before scaling to Spark or cloud platforms.

Related conversions

Frequent conversions across the catalogue:

• JPG to PNG
• MP4 to MP3
• PDF to DOCX
• PNG to JPG
• MP3 to WAV
• DOCX to PDF