1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Python
pip install duckdb

# R
install.packages("duckdb")

# CLI
curl https://install.duckdb.org | sh
# or: brew install duckdb

# Node.js
npm install duckdb

# Go
go get github.com/marcboeker/go-duckdb

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- In-memory database: just connect and query
import duckdb
con = duckdb.connect()  # or duckdb.connect('mydb.duckdb') for persistence

-- Create tables
CREATE TABLE orders AS SELECT * FROM read_parquet('orders/*.parquet');

-- DuckDB's GROUP BY shorthand: refer to SELECT columns by position or alias
SELECT
    date_trunc('month', created_at) AS month,
    status,
    COUNT(*) AS order_count,
    SUM(amount_cents) / 100.0 AS revenue
FROM orders
GROUP BY ALL   -- ← GROUP BY ALL: groups by all non-aggregate columns automatically
ORDER BY 1, 2;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

-- COLUMNS expression: apply a function to multiple columns at once
SELECT
    order_id,
    COLUMNS('amount.*')  -- matches all columns starting with 'amount'
FROM orders;

-- EXCLUDE and REPLACE in SELECT
SELECT * EXCLUDE (internal_id, raw_payload) FROM orders;

SELECT * REPLACE (amount_cents / 100.0 AS amount_dollars) FROM orders;

-- Struct and list literals
SELECT {'name': 'Alice', 'age': 30} AS person;
SELECT [1, 2, 3] AS numbers;

-- Unnesting arrays
SELECT unnest([1, 2, 3]) AS value;

-- PIVOT
PIVOT orders ON status USING SUM(amount_cents);
-- Produces columns: pending_sum, shipped_sum, cancelled_sum, etc.

-- UNPIVOT
UNPIVOT wide_table ON COLUMNS(* EXCLUDE id) INTO NAME metric VALUE amount;

-- String formatting
SELECT format('Order #{} placed on {}', order_id, created_at::DATE) AS summary
FROM orders LIMIT 5;

-- List comprehension-style expressions
SELECT list_transform([1, 2, 3, 4, 5], x -> x * x) AS squares;

-- Chained method-style string functions
SELECT 'hello world'.upper().replace(' ', '_') AS slug;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

-- Named windows reduce repetition
SELECT
    order_id,
    customer_id,
    amount_cents,
    SUM(amount_cents)  OVER customer_window AS customer_total,
    AVG(amount_cents)  OVER customer_window AS customer_avg,
    ROW_NUMBER()       OVER customer_window AS order_rank,
    LAG(amount_cents)  OVER customer_window AS prev_order_amount
FROM orders
WINDOW customer_window AS (
    PARTITION BY customer_id
    ORDER BY created_at
    ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
);

-- Quantile window (no standard SQL equivalent)
SELECT
    order_id,
    QUANTILE_CONT(amount_cents, 0.5) OVER (PARTITION BY category) AS median_by_category,
    QUANTILE_CONT(amount_cents, [0.25, 0.5, 0.75]) OVER () AS quartiles
FROM orders;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

-- Statistical aggregates
SELECT
    AVG(amount_cents)                     AS mean,
    STDDEV(amount_cents)                  AS std_dev,
    MEDIAN(amount_cents)                  AS median,
    APPROX_QUANTILE(amount_cents, 0.99)   AS p99,
    KURTOSIS(amount_cents)                AS kurtosis,
    CORR(amount_cents, item_count)        AS amount_item_correlation
FROM orders;

-- List aggregation
SELECT
    customer_id,
    LIST(order_id ORDER BY created_at) AS order_history,
    LIST(status)[-1] AS latest_status  -- Last element of the list
FROM orders
GROUP BY customer_id;

-- Histogram
SELECT histogram(amount_cents) FROM orders;
-- Returns a struct with bin edges and counts

-- Mode
SELECT MODE(status) AS most_common_status FROM orders;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

-- Single file
SELECT COUNT(*) FROM read_parquet('data/orders.parquet');

-- Multiple files with glob
SELECT * FROM read_parquet('data/orders/year=2025/**/*.parquet');

-- Hive partitioning: extract partition columns from path
SELECT year, month, COUNT(*)
FROM read_parquet(
    'data/orders/year=*/month=*/*.parquet',
    hive_partitioning = true   -- Adds year and month as columns
)
GROUP BY ALL;

-- Create a view over Parquet files (no data loaded)
CREATE VIEW orders AS SELECT * FROM read_parquet('data/orders/**/*.parquet');

-- Schema inspection
DESCRIBE SELECT * FROM read_parquet('data/orders.parquet');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

-- CSV with type inference
SELECT * FROM read_csv('data/sales.csv');

-- CSV with explicit schema
SELECT * FROM read_csv('data/sales.csv',
    columns = {
        'order_id': 'BIGINT',
        'amount': 'DECIMAL(10,2)',
        'created_at': 'TIMESTAMP'
    },
    dateformat = '%Y-%m-%d',
    null_padding = true
);

-- CSV with auto-detect (examines first 10k rows)
SELECT * FROM read_csv_auto('data/sales.csv');

-- NDJSON (newline-delimited JSON)
SELECT * FROM read_json('data/events.jsonl');

-- JSON with schema
SELECT
    json_extract(data, '$.user.id')::BIGINT AS user_id,
    json_extract_string(data, '$.event_type') AS event_type
FROM read_json('data/events.jsonl', columns = {data: 'JSON'});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

-- Install and load the httpfs extension (auto-loaded in Python package)
INSTALL httpfs;
LOAD httpfs;

-- Configure AWS credentials
SET s3_region = 'us-east-1';
SET s3_access_key_id = 'AKIA...';
SET s3_secret_access_key = '...';
-- Or use instance metadata / credential chain:
SET s3_use_credential_chain = true;

-- Query S3 directly — no download step
SELECT
    date_trunc('day', event_time) AS day,
    event_type,
    COUNT(*) AS events
FROM read_parquet('s3://my-data-lake/events/year=2026/month=03/**/*.parquet',
    hive_partitioning = true)
WHERE year = 2026 AND month = 3
GROUP BY ALL
ORDER BY 1, 3 DESC;

-- GCS
SET gcs_token = '...';
SELECT * FROM read_parquet('gcs://my-bucket/data/*.parquet');

-- Azure Blob Storage
SET azure_storage_connection_string = 'DefaultEndpointsProtocol=https;...';
SELECT * FROM read_parquet('azure://mycontainer/data/*.parquet');

1
2
3
4
5
6
7
8
9

-- Delta Lake (requires delta extension)
INSTALL delta;
LOAD delta;
SELECT * FROM delta_scan('s3://my-data-lake/delta/orders/');

-- Apache Iceberg (requires iceberg extension)
INSTALL iceberg;
LOAD iceberg;
SELECT * FROM iceberg_scan('s3://my-data-lake/iceberg/orders/');

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import duckdb
import pandas as pd

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

# Persistent database
con = duckdb.connect('analytics.duckdb')

# Execute and fetch as various types
df = con.execute("SELECT * FROM orders LIMIT 1000").df()           # Pandas DataFrame
arrow = con.execute("SELECT * FROM orders").arrow()                # PyArrow Table
polars = con.execute("SELECT * FROM orders").pl()                  # Polars DataFrame
tuples = con.execute("SELECT * FROM orders").fetchall()            # List of tuples
one = con.execute("SELECT COUNT(*) FROM orders").fetchone()[0]     # Single value

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import duckdb
import pandas as pd

# DuckDB can query Pandas DataFrames in-place
orders_df = pd.read_parquet('data/orders.parquet')
customers_df = pd.read_parquet('data/customers.parquet')

# Reference DataFrames by variable name in SQL
result = duckdb.execute("""
    SELECT
        c.name,
        c.email,
        COUNT(o.order_id) AS order_count,
        SUM(o.amount_cents) / 100.0 AS total_spend
    FROM orders_df o
    JOIN customers_df c ON o.customer_id = c.id
    WHERE o.created_at >= '2026-01-01'
    GROUP BY ALL
    ORDER BY total_spend DESC
    LIMIT 20
""").df()

# Register with an explicit name
con.register('orders', orders_df)
con.register('customers', customers_df)

# Or use the relation API (method chaining)
orders_rel = con.table('orders')
result = (
    orders_rel
    .filter("status = 'shipped'")
    .aggregate("customer_id, SUM(amount_cents) AS total", "customer_id")
    .order("total DESC")
    .limit(10)
    .df()
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq

# Read large Parquet file — DuckDB scans it without loading to Python memory
arrow_table = duckdb.execute("""
    SELECT
        date_trunc('week', created_at) AS week,
        SUM(amount_cents) AS weekly_revenue
    FROM read_parquet('s3://my-bucket/orders/**/*.parquet')
    WHERE created_at >= '2026-01-01'
    GROUP BY ALL
    ORDER BY 1
""").arrow()

# The result is an Arrow table — zero-copy to Pandas
df = arrow_table.to_pandas()

# Or write directly to Parquet
pq.write_table(arrow_table, 'weekly_revenue.parquet')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

#!/usr/bin/env python3
"""
Transform raw event logs into an aggregated user activity summary.
Replaces a Spark job that took 45 minutes on a 10-node cluster.
DuckDB does this in 4 minutes on a single machine.

Input: 80GB of compressed Parquet event logs on S3
Output: user_activity_summary.parquet (~500MB)
"""
import duckdb
from pathlib import Path

def run_transform(
    input_s3_path: str,
    output_path: str,
    start_date: str,
    end_date: str,
):
    con = duckdb.connect()
    con.execute("SET s3_use_credential_chain = true")
    con.execute("SET s3_region = 'us-east-1'")

    # Configure DuckDB for large workloads
    con.execute("SET memory_limit = '24GB'")
    con.execute("SET threads = 16")
    con.execute("SET temp_directory = '/tmp/duckdb'")

    con.execute(f"""
        COPY (
            WITH raw_events AS (
                SELECT
                    user_id,
                    event_type,
                    DATE_TRUNC('day', event_time) AS event_date,
                    session_id,
                    properties
            FROM read_parquet('{input_s3_path}',
                hive_partitioning = true,
                filename = true)
            WHERE event_date BETWEEN '{start_date}' AND '{end_date}'
                AND user_id IS NOT NULL
            ),
            session_metrics AS (
                SELECT
                    user_id,
                    session_id,
                    event_date,
                    MIN(event_time) AS session_start,
                    MAX(event_time) AS session_end,
                    COUNT(*) AS events_in_session,
                    epoch_ms(MAX(event_time) - MIN(event_time)) / 1000.0
                        AS session_duration_seconds
                FROM raw_events
                GROUP BY 1, 2, 3
            )
            SELECT
                r.user_id,
                r.event_date,
                COUNT(DISTINCT r.session_id) AS sessions,
                COUNT(*) AS total_events,
                COUNT(*) FILTER (WHERE r.event_type = 'purchase') AS purchases,
                SUM(TRY_CAST(r.properties->>'$.amount' AS DECIMAL(10,2)))
                    FILTER (WHERE r.event_type = 'purchase') AS purchase_revenue,
                AVG(s.session_duration_seconds) AS avg_session_duration_secs,
                FIRST(r.properties->>'$.platform') AS platform,
                LIST(DISTINCT r.event_type) AS event_types_seen
            FROM raw_events r
            JOIN session_metrics s USING (user_id, session_id, event_date)
            GROUP BY 1, 2
        )
        TO '{output_path}'
        (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
    """)

    print("Transform complete.")


if __name__ == "__main__":
    run_transform(
        input_s3_path="s3://my-data-lake/events/year=2026/month=03/**/*.parquet",
        output_path="s3://my-data-lake/aggregated/user_activity_2026_03.parquet",
        start_date="2026-03-01",
        end_date="2026-03-31",
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- Check current settings
SELECT * FROM duckdb_settings() WHERE name IN (
    'memory_limit', 'threads', 'temp_directory', 'worker_threads'
);

-- Configure for large analytical workloads
SET memory_limit = '80%';          -- Use 80% of available RAM
SET threads = 16;                   -- Match physical core count (not hyperthreads)
SET temp_directory = '/nvme/tmp';   -- Fast local SSD for spill-to-disk
SET enable_progress_bar = true;     -- Show progress for long queries

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

-- Enable profiling
PRAGMA enable_profiling;
PRAGMA profiling_output = '/tmp/profile.json';

-- Run your query
SELECT customer_id, SUM(amount_cents)
FROM orders
WHERE created_at >= '2026-01-01'
GROUP BY customer_id;

-- Read the profile
SELECT * FROM read_json('/tmp/profile.json');

-- Or use EXPLAIN ANALYZE
EXPLAIN ANALYZE
SELECT customer_id, SUM(amount_cents)
FROM orders
WHERE created_at >= '2026-01-01'
GROUP BY customer_id;

-- EXPLAIN shows the physical plan
EXPLAIN SELECT ...;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import duckdb

# Write a sorted Parquet file — dramatically improves filter performance
duckdb.execute("""
    COPY (
        SELECT * FROM read_parquet('s3://bucket/orders/**/*.parquet')
        ORDER BY created_at, customer_id  -- Sort by most-filtered columns
    )
    TO 's3://bucket/orders_sorted.parquet'
    (FORMAT PARQUET, ROW_GROUP_SIZE 100000, COMPRESSION ZSTD)
""")

1
2
3
4
5

CREATE TABLE customers AS SELECT * FROM read_parquet('customers.parquet');
CREATE INDEX idx_customers_email ON customers(email);

-- Point lookup now uses the index
SELECT * FROM customers WHERE email = 'alice@example.com';

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- With hive_partitioning=true and a WHERE clause on partition columns,
-- DuckDB reads only the relevant directories
SELECT COUNT(*)
FROM read_parquet(
    's3://bucket/events/year=*/month=*/day=*/*.parquet',
    hive_partitioning = true
)
WHERE year = 2026 AND month = 3;
-- Only reads s3://bucket/events/year=2026/month=3/**
-- Ignores all other year/month partitions entirely

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- Force a specific join algorithm
SELECT /*+ HASH_JOIN(orders, customers) */ ...
SELECT /*+ MERGE_JOIN(orders, customers) */ ...

-- For large joins where one side fits in memory, broadcast join is automatic
-- For very large joins, DuckDB spills to disk automatically

-- Check what DuckDB chose
EXPLAIN SELECT o.*, c.name
FROM orders o JOIN customers c ON o.customer_id = c.id;

1

pip install dbt-duckdb

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# profiles.yml
jaffle_shop:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: 'jaffle_shop.duckdb'
      threads: 4
      # Attach external Parquet files as schemas
      external_root: 'data/'

    prod:
      type: duckdb
      path: 's3://my-bucket/warehouse/prod.duckdb'
      s3_region: us-east-1
      s3_use_credential_chain: true
      threads: 16

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- models/staging/stg_orders.sql
-- DuckDB model reading raw Parquet from S3
{{ config(materialized='view') }}

SELECT
    order_id,
    customer_id,
    status,
    amount_cents / 100.0 AS amount,
    created_at::DATE AS order_date
FROM read_parquet('{{ env_var("RAW_DATA_PATH") }}/orders/**/*.parquet')
WHERE created_at >= '{{ var("start_date", "2026-01-01") }}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

-- models/marts/customer_lifetime_value.sql
{{ config(materialized='table') }}

WITH order_stats AS (
    SELECT
        customer_id,
        COUNT(*) AS order_count,
        SUM(amount) AS total_spent,
        MIN(order_date) AS first_order_date,
        MAX(order_date) AS last_order_date,
        MAX(order_date) - MIN(order_date) AS customer_age_days
    FROM {{ ref('stg_orders') }}
    WHERE status = 'shipped'
    GROUP BY customer_id
)
SELECT
    c.customer_id,
    c.email,
    o.order_count,
    o.total_spent,
    o.total_spent / NULLIF(o.customer_age_days, 0) * 365 AS annualized_ltv,
    CASE
        WHEN o.total_spent > 1000 THEN 'high'
        WHEN o.total_spent > 200  THEN 'medium'
        ELSE 'low'
    END AS value_tier
FROM {{ ref('stg_customers') }} c
LEFT JOIN order_stats o USING (customer_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- Generate a complete date spine and fill gaps with 0
WITH date_spine AS (
    SELECT UNNEST(generate_series(
        DATE '2026-01-01',
        DATE '2026-03-31',
        INTERVAL 1 DAY
    ))::DATE AS date
),
daily_revenue AS (
    SELECT
        created_at::DATE AS date,
        SUM(amount_cents) / 100.0 AS revenue
    FROM orders
    GROUP BY 1
)
SELECT
    d.date,
    COALESCE(r.revenue, 0) AS revenue
FROM date_spine d
LEFT JOIN daily_revenue r USING (date)
ORDER BY 1;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

-- Assign session IDs: new session if > 30 minutes since last event
WITH events_with_gap AS (
    SELECT
        user_id,
        event_time,
        event_type,
        LAG(event_time) OVER (
            PARTITION BY user_id ORDER BY event_time
        ) AS prev_event_time
    FROM events
),
session_boundaries AS (
    SELECT
        *,
        CASE
            WHEN prev_event_time IS NULL
              OR epoch_ms(event_time - prev_event_time) > 1800000
            THEN 1 ELSE 0
        END AS is_new_session
    FROM events_with_gap
)
SELECT
    user_id,
    event_time,
    event_type,
    SUM(is_new_session) OVER (
        PARTITION BY user_id ORDER BY event_time
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS session_number
FROM session_boundaries;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- E-commerce funnel: view → add_to_cart → checkout → purchase
WITH funnel_steps AS (
    SELECT user_id,
        MAX(CASE WHEN event_type = 'product_view'    THEN 1 END) AS step1,
        MAX(CASE WHEN event_type = 'add_to_cart'     THEN 1 END) AS step2,
        MAX(CASE WHEN event_type = 'checkout_start'  THEN 1 END) AS step3,
        MAX(CASE WHEN event_type = 'purchase'        THEN 1 END) AS step4
    FROM events
    WHERE event_date BETWEEN '2026-03-01' AND '2026-03-31'
    GROUP BY user_id
)
SELECT
    COUNT(*) FILTER (WHERE step1 = 1) AS viewed_product,
    COUNT(*) FILTER (WHERE step2 = 1) AS added_to_cart,
    COUNT(*) FILTER (WHERE step3 = 1) AS started_checkout,
    COUNT(*) FILTER (WHERE step4 = 1) AS purchased,
    ROUND(100.0 * COUNT(*) FILTER (WHERE step2 = 1) /
          NULLIF(COUNT(*) FILTER (WHERE step1 = 1), 0), 1) AS view_to_cart_pct,
    ROUND(100.0 * COUNT(*) FILTER (WHERE step4 = 1) /
          NULLIF(COUNT(*) FILTER (WHERE step1 = 1), 0), 1) AS overall_conversion_pct
FROM funnel_steps;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

-- Export to Parquet (columnar, compressed)
COPY orders TO 'output/orders.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);

-- Export to CSV
COPY (SELECT * FROM orders WHERE status = 'shipped')
TO 'output/shipped_orders.csv' (HEADER, DELIMITER ',');

-- Export partitioned by date
COPY orders TO 'output/orders/'
(FORMAT PARQUET, PARTITION_BY (order_date), OVERWRITE_OR_IGNORE);
-- Creates: output/orders/order_date=2026-03-01/data.parquet, etc.

-- Export directly to S3
COPY orders TO 's3://my-bucket/exports/orders.parquet' (FORMAT PARQUET);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import duckdb

# Connect to MotherDuck (requires token)
con = duckdb.connect('md:?motherduck_token=<your_token>')

# Hybrid execution: local data + cloud catalog
con.execute("""
    -- Query local file and cloud table in the same SQL statement
    SELECT l.order_id, c.customer_name
    FROM read_parquet('/local/orders.parquet') l
    JOIN md:prod.customers c ON l.customer_id = c.id
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

-- File reading
SELECT * FROM read_parquet('file.parquet');
SELECT * FROM read_csv_auto('file.csv');
SELECT * FROM read_json('file.jsonl');
SELECT * FROM read_parquet('s3://bucket/path/**/*.parquet', hive_partitioning=true);

-- Settings
SET memory_limit = '16GB';
SET threads = 8;
SET temp_directory = '/tmp/duckdb';
SHOW ALL TABLES;
DESCRIBE my_table;

-- DuckDB-specific SQL
SELECT * EXCLUDE (col1, col2) FROM t;
SELECT * REPLACE (expr AS col) FROM t;
GROUP BY ALL;
PIVOT t ON col USING SUM(val);
UNPIVOT t ON COLUMNS(*) INTO NAME k VALUE v;

-- Export
COPY t TO 'out.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);
COPY t TO 'out/' (FORMAT PARQUET, PARTITION_BY (date_col));

-- Profiling
PRAGMA enable_profiling;
EXPLAIN ANALYZE SELECT ...;