An Implementation Guide to Building a DuckDB-Python Analytics Pipeline with SQL, DataFrames, Parquet, UDFs, and Performance Profiling

import subprocess, sys


for pkg in ["duckdb", "pandas", "pyarrow", "polars"]:
   strive:
       subprocess.check_call(
           [sys.executable, "-m", "pip", "install", "-q", pkg],
           stderr=subprocess.DEVNULL,
       )
   besides subprocess.CalledProcessError:
       subprocess.check_call(
           [sys.executable, "-m", "pip", "install", "-q", "--break-system-packages", pkg],
           stderr=subprocess.DEVNULL,
       )


import duckdb
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import polars as pl
import numpy as np
import json, os, time, threading, tempfile
from datetime import date, datetime, timedelta


print(f"DuckDB model : {duckdb.__version__}")
print(f"Pandas model : {pd.__version__}")
print(f"PyArrow model: {pa.__version__}")
print(f"Polars model : {pl.__version__}")
print("=" * 72)


WORKDIR = tempfile.mkdtemp(prefix="duckdb_tutorial_")
os.chdir(WORKDIR)
print(f"Working listing: {WORKDIR}n")


print("=" * 72)
print("SECTION 1: Connection Management")
print("=" * 72)


con = duckdb.join()
print(con.sql("SELECT 'Hello from in-memory DuckDB!' AS greeting").fetchone()[0])


DB_PATH = os.path.be a part of(WORKDIR, "tutorial.duckdb")
pcon = duckdb.join(DB_PATH)
pcon.sql("CREATE OR REPLACE TABLE continued(id INT, val TEXT)")
pcon.sql("INSERT INTO continued VALUES (1,'alpha'), (2,'beta')")
print("Persisted rows:", pcon.sql("SELECT rely(*) FROM continued").fetchone()[0])
pcon.shut()


pcon2 = duckdb.join(DB_PATH)
print("After re-open :", pcon2.sql("SELECT * FROM continued ORDER BY id").fetchall())
pcon2.shut()


con_cfg = duckdb.join(config={
   "threads": 2,
   "memory_limit": "512MB",
   "default_order": "DESC",
})
print("Configured threads:", con_cfg.sql("SELECT current_setting('threads')").fetchone()[0])
con_cfg.shut()


with duckdb.join() as ctx:
   ctx.sql("SELECT 42 AS reply").present()


print()


print("=" * 72)
print("SECTION 2: Synthetic Data Generation")
print("=" * 72)


con = duckdb.join()


con.sql("""
   CREATE OR REPLACE TABLE gross sales AS
   SELECT
       i                                                     AS order_id,
       '2023-01-01'::DATE + (i % 365)::INT                  AS order_date,
       CASE (i % 5)
           WHEN 0 THEN 'Electronics'
           WHEN 1 THEN 'Clothing'
           WHEN 2 THEN 'Groceries'
           WHEN 3 THEN 'Furniture'
           ELSE        'Books'
       END                                                   AS class,
       ROUND(10 + random() * 990, 2)                         AS quantity,
       CASE (i % 3)
           WHEN 0 THEN 'US'
           WHEN 1 THEN 'EU'
           ELSE        'APAC'
       END                                                   AS area,
       CASE WHEN random() < 0.1 THEN TRUE ELSE FALSE END    AS returned
   FROM generate_series(1, 100000) t(i)
""")


con.sql("SUMMARIZE gross sales").present()
print()


print("=" * 72)
print("SECTION 3: Zero-Copy DataBody Integration")
print("=" * 72)


pdf = pd.DataBody({
   "product": ["Widget", "Gadget", "Doohickey", "Thingamajig"],
   "worth":   [9.99, 24.50, 4.75, 15.00],
   "inventory":   [120, 45, 300, 78],
})
print("Query Pandas DF immediately:")
con.sql("SELECT product, worth * inventory AS inventory_value FROM pdf ORDER BY inventory_value DESC").present()


plf = pl.DataBody({
   "metropolis":   ["Montreal", "Toronto", "Vancouver", "Calgary"],
   "temp_c": [-12.5, -5.0, 3.0, -18.0],
})
print("Query Polars DF immediately:")
con.sql("SELECT metropolis, temp_c, temp_c * 9/5 + 32 AS temp_f FROM plf WHERE temp_c < 0").present()


arrow_tbl = pa.desk({
   "sensor_id": [1, 2, 3, 4, 5],
   "studying":   [23.1, 47.8, 12.3, 99.0, 55.5],
})
print("Query Arrow desk immediately:")
con.sql("SELECT sensor_id, studying FROM arrow_tbl WHERE studying > 30").present()


print()


print("=" * 72)
print("SECTION 4: Result Conversion")
print("=" * 72)


q = "SELECT class, SUM(quantity) AS whole FROM gross sales GROUP BY class ORDER BY whole DESC"


print("→ Python checklist  :", con.sql(q).fetchall()[:2], "...")
print("→ Pandas DF    :n", con.sql(q).df().head(3))
print("→ Polars DF    :n", con.sql(q).pl().head(3))
print("→ Arrow Table  :", con.sql(q).arrow().schema)
print("→ NumPy arrays :", {okay: v[:2] for okay, v in con.sql(q).fetchnumpy().gadgets()})
print()

print("=" * 72)
print("SECTION 5: Relational API")
print("=" * 72)


rel = (
   con.desk("gross sales")
      .filter("NOT returned")
      .mixture("class, area, SUM(quantity) AS income, COUNT(*) AS orders")
      .filter("income > 1000000")
      .order("income DESC")
      .restrict(10)
)
print("Relational API end result:")
rel.present()


top_cats = con.sql("SELECT DISTINCT class FROM gross sales ORDER BY class LIMIT 3")
print("Top classes relation fed into subsequent question:")
con.sql("SELECT s.* FROM gross sales s SEMI JOIN top_cats ON s.class = top_cats.class LIMIT 5").present()


print()


print("=" * 72)
print("SECTION 6: Window Functions & Advanced SQL")
print("=" * 72)


con.sql("""
   WITH each day AS (
       SELECT
           order_date,
           area,
           SUM(quantity) AS daily_rev
       FROM gross sales
       WHERE NOT returned
       GROUP BY order_date, area
   )
   SELECT
       order_date,
       area,
       daily_rev,
       SUM(daily_rev) OVER (
           PARTITION BY area ORDER BY order_date
       ) AS cum_revenue,
       AVG(daily_rev) OVER (
           PARTITION BY area ORDER BY order_date
           ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
       ) AS rolling_7d_avg
   FROM each day
   QUALIFY row_number() OVER (PARTITION BY area ORDER BY order_date DESC) <= 3
   ORDER BY area, order_date DESC
""").present()


print("PIVOT desk:")
con.sql("""
   PIVOT gross sales
   ON area
   USING SUM(quantity)
   GROUP BY class
   ORDER BY class
""").present()


print()


print("=" * 72)
print("SECTION 7: Complex / Nested Types")
print("=" * 72)


con.sql("""
   CREATE OR REPLACE TABLE customers AS
   SELECT
       i AS user_id,
        i::TEXT, 'final': 'Surname_'   AS title,
       [i * 10, i * 20, i * 30]                                              AS scores,
       MAP {'tier': CASE WHEN i % 2 = 0 THEN 'gold' ELSE 'silver' END,
            'area': CASE WHEN i % 3 = 0 THEN 'US' ELSE 'EU' END}          AS metadata
   FROM generate_series(1, 5) t(i)
""")


print("Struct discipline entry, checklist indexing, map extraction:")
con.sql("""
   SELECT
       user_id,
       title.first                  AS first_name,
       scores[1]                   AS first_score,
       list_aggregate(scores,'sum') AS total_score,
       metadata['tier']            AS tier
   FROM customers
""").present()


print("Unnesting a checklist column:")
con.sql("""
   SELECT user_id, unnest(scores) AS individual_score
   FROM customers
   WHERE user_id <= 3
""").present()


print()


print("=" * 72)
print("SECTION 8: Python UDFs")
print("=" * 72)


def celsius_to_fahrenheit(c):
   return c * 9 / 5 + 32


con.create_function("c2f", celsius_to_fahrenheit, ["DOUBLE"], "DOUBLE")
con.sql("SELECT metropolis, temp_c, c2f(temp_c) AS temp_f FROM plf").present()


import pyarrow.compute as computer


def vectorized_discount(costs):
   """Apply a 15% low cost to all costs."""
   return computer.multiply(costs, 0.85)


con.create_function(
   "low cost",
   vectorized_discount,
   ["DOUBLE"],
   "DOUBLE",
   sort="arrow",
)
print("Vectorized UDF (low cost):")
con.sql("SELECT product, worth, low cost(worth) AS sale_price FROM pdf").present()


print()


print("=" * 72)
print("SECTION 9: File I/O")
print("=" * 72)


con.sql("COPY (SELECT * FROM gross sales LIMIT 1000) TO 'sales_sample.csv'  (HEADER, DELIMITER ',')")
con.sql("COPY (SELECT * FROM gross sales LIMIT 1000) TO 'sales_sample.parquet' (FORMAT PARQUET)")
con.sql("COPY (SELECT * FROM gross sales LIMIT 100)  TO 'sales_sample.json'    (FORMAT JSON, ARRAY true)")


print("Files written:", [f for f in os.listdir('.') if 'sales_sample' in f])


print("nCSV row rely     :", con.sql("SELECT rely(*) FROM 'sales_sample.csv'").fetchone()[0])
print("Parquet row rely :", con.sql("SELECT rely(*) FROM 'sales_sample.parquet'").fetchone()[0])
print("JSON row rely    :", con.sql("SELECT rely(*) FROM 'sales_sample.json'").fetchone()[0])


print("nParquet file metadata:")
con.sql("SELECT * FROM parquet_metadata('sales_sample.parquet')").present()


print()

print("=" * 72)
print("SECTION 10: Hive-Partitioned Parquet")
print("=" * 72)


con.sql("""
   COPY gross sales TO 'partitioned_data' (
       FORMAT PARQUET,
       PARTITION_BY (area, class),
       OVERWRITE_OR_IGNORE
   )
""")
print("Partitioned listing tree (first 15 gadgets):")
for i, (root, dirs, recordsdata) in enumerate(os.stroll("partitioned_data")):
   for f in recordsdata:
       fp = os.path.be a part of(root, f)
       print(f"  {fp}")
   if i > 15:
       print("  ...")
       break


print("nReading partitioned information again (US + Electronics solely):")
con.sql("""
   SELECT rely(*) AS cnt, MIN(quantity), MAX(quantity)
   FROM read_parquet('partitioned_data/**/*.parquet', hive_partitioning = true)
   WHERE area = 'US' AND class = 'Electronics'
""").present()


print()


print("=" * 72)
print("SECTION 11: Prepared Statements")
print("=" * 72)


end result = con.execute(
   "SELECT * FROM gross sales WHERE class = $1 AND quantity > $2 LIMIT 5",
   ["Electronics", 500.0]
).fetchdf()
print("Parameterized question end result:n", end result)


con.sql("SET VARIABLE target_region = 'EU'")
con.sql("""
   SELECT class, AVG(quantity) AS avg_amt
   FROM gross sales
   WHERE area = getvariable('target_region')
   GROUP BY class
   ORDER BY avg_amt DESC
""").present()


print()


print("=" * 72)
print("SECTION 12: Transactions")
print("=" * 72)


con.sql("CREATE OR REPLACE TABLE accounts(id INT, stability DOUBLE)")
con.sql("INSERT INTO accounts VALUES (1, 1000), (2, 500)")


con.start()
strive:
   con.sql("UPDATE accounts SET stability = stability - 200 WHERE id = 1")
   con.sql("UPDATE accounts SET stability = stability + 200 WHERE id = 2")
   con.commit()
   print("Transaction dedicated efficiently")
besides Exception as e:
   con.rollback()
   print(f"Transaction rolled again: {e}")


con.sql("SELECT * FROM accounts").present()
print()


print("=" * 72)
print("SECTION 13: Appender (Bulk Insert)")
print("=" * 72)


con.sql("CREATE OR REPLACE TABLE fast_load(id INT, title TEXT, worth DOUBLE)")


bulk_df = pd.DataBody({
   "id":    vary(50_000),
   "title":  [f"item_{i}" for i in range(50_000)],
   "worth": [float(i) * 1.1 for i in range(50_000)],
})


begin = time.perf_counter()
con.append("fast_load", bulk_df)
elapsed = time.perf_counter() - begin


print(f"Bulk-inserted {con.sql('SELECT rely(*) FROM fast_load').fetchone()[0]:,} rows in {elapsed:.4f}s")
con.sql("SELECT * FROM fast_load LIMIT 5").present()
print()


print("=" * 72)
print("SECTION 14: Replacement Scans (question any Python var)")
print("=" * 72)


my_dict_data = pd.DataBody({
   "title": ["Alice", "Bob", "Charlie"],
   "age":  [30, 25, 35],
})
con.sql("SELECT * FROM my_dict_data WHERE age > 28").present()
print()

print("=" * 72)
print("SECTION 15: SQL Macros")
print("=" * 72)


con.sql("""
   CREATE OR REPLACE MACRO revenue_tier(amt) AS
       CASE
           WHEN amt > 800 THEN 'High'
           WHEN amt > 400 THEN 'Medium'
           ELSE                'Low'
       END
""")
print("Scalar macro:")
con.sql("SELECT class, quantity, revenue_tier(quantity) AS tier FROM gross sales LIMIT 8").present()


con.sql("""
   CREATE OR REPLACE MACRO top_by_category(cat, n) AS TABLE
       SELECT * FROM gross sales WHERE class = cat ORDER BY quantity DESC LIMIT n
""")
print("Table macro — high 3 Electronics:")
con.sql("SELECT * FROM top_by_category('Electronics', 3)").present()
print()


print("=" * 72)
print("SECTION 16: Recursive CTE")
print("=" * 72)


con.sql("""
   CREATE OR REPLACE TABLE workers(id INT, title TEXT, manager_id INT);
   INSERT INTO workers VALUES
       (1, 'CEO',     NULL),
       (2, 'VP Eng',  1),
       (3, 'VP Sales',1),
       (4, 'Sr Eng',  2),
       (5, 'Jr Eng',  4),
       (6, 'Sales Rep', 3);
""")


print("Org chart by way of recursive CTE:")
con.sql("""
   WITH RECURSIVE org AS (
       SELECT id, title, manager_id, 0 AS depth, title AS path
       FROM workers WHERE manager_id IS NULL
       UNION ALL
       SELECT e.id, e.title, e.manager_id, o.depth + 1,
              o.path || ' → ' || e.title
       FROM workers e JOIN org o ON e.manager_id = o.id
   )
   SELECT repeat('  ', depth) || title AS hierarchy, path
   FROM org
   ORDER BY path
""").present()
print()


print("=" * 72)
print("SECTION 17: Full-Text Search (FTS)")
print("=" * 72)


strive:
   con.install_extension("fts")
   con.load_extension("fts")


   con.sql("""
       CREATE OR REPLACE TABLE paperwork(id INT, physique TEXT);
       INSERT INTO paperwork VALUES
           (1, 'DuckDB is a quick in-process analytical database'),
           (2, 'Python integration permits querying Pandas DataFrames'),
           (3, 'Parquet recordsdata could be learn immediately with out loading'),
           (4, 'Window capabilities and CTEs make complicated analytics straightforward'),
           (5, 'The columnar engine processes information blazingly quick');
   """)


   con.sql("PRAGMA create_fts_index('paperwork', 'id', 'physique', stemmer='english')")


   print("FTS seek for 'quick analytical':")
   con.sql("""
       SELECT id, physique, fts_main_documents.match_bm25(id, 'quick analytical') AS rating
       FROM paperwork
       WHERE rating IS NOT NULL
       ORDER BY rating DESC
   """).present()
besides Exception as e:
   print(f"(Skipped — FTS extension not out there: {e})")
print()


print("=" * 72)
print("SECTION 18: AsOf Joins")
print("=" * 72)


con.sql("""
   CREATE OR REPLACE TABLE stock_prices AS
   SELECT * FROM (VALUES
       ('2024-01-01 09:30'::TIMESTAMP, 'AAPL', 150.0),
       ('2024-01-01 10:00'::TIMESTAMP, 'AAPL', 152.5),
       ('2024-01-01 10:30'::TIMESTAMP, 'AAPL', 151.0),
       ('2024-01-01 11:00'::TIMESTAMP, 'AAPL', 153.0)
   ) AS t(ts, ticker, worth);


   CREATE OR REPLACE TABLE trades AS
   SELECT * FROM (VALUES
       ('2024-01-01 09:45'::TIMESTAMP, 'AAPL', 100),
       ('2024-01-01 10:15'::TIMESTAMP, 'AAPL', 200),
       ('2024-01-01 10:50'::TIMESTAMP, 'AAPL', 150)
   ) AS t(trade_ts, ticker, shares);
""")


print("AsOf Join — match every commerce to the newest worth:")
con.sql("""
   SELECT
       t.trade_ts,
       t.shares,
       p.ts   AS price_ts,
       p.worth,
       t.shares * p.worth AS trade_value
   FROM trades t
   ASOF JOIN stock_prices p
       ON t.ticker = p.ticker AND t.trade_ts >= p.ts
""").present()
print()

print("=" * 72)
print("SECTION 19: Query Profiling")
print("=" * 72)


print("EXPLAIN output:")
con.sql("EXPLAIN SELECT class, SUM(quantity) FROM gross sales GROUP BY class").present()


con.sql("PRAGMA enable_profiling = 'json'")
con.sql("PRAGMA profiling_output = 'profile.json'")
con.sql("SELECT class, AVG(quantity), STDDEV(quantity) FROM gross sales GROUP BY class")
con.sql("PRAGMA disable_profiling")


if os.path.exists("profile.json"):
   with open("profile.json") as f:
       profile = json.load(f)
   print(f"Query timing: {profile.get('timing', 'N/A')}s")


print()


print("=" * 72)
print("SECTION 20: Multi-Threaded Queries")
print("=" * 72)


outcomes = {}


def employee(thread_id, db_path=None):
   """Each thread creates its personal connection."""
   local_con = duckdb.join()
   local_con.sql(f"""
       CREATE TABLE t AS SELECT i, i*{thread_id} AS val
       FROM generate_series(1, 10000) t(i)
   """)
   whole = local_con.sql("SELECT SUM(val) FROM t").fetchone()[0]
   outcomes[thread_id] = whole
   local_con.shut()


threads = [threading.Thread(target=worker, args=(tid,)) for tid in range(4)]
for t in threads:
   t.begin()
for t in threads:
   t.be a part of()


print("Thread outcomes:", outcomes)
print()


print("=" * 72)
print("SECTION 21: Quick Benchmark — DuckDB vs Pandas")
print("=" * 72)


N = 1_000_000
big_df = pd.DataBody({
   "group": np.random.selection(["A","B","C","D","E"], N),
   "worth": np.random.randn(N) * 100,
   "ts":    pd.date_range("2020-01-01", intervals=N, freq="s"),
})


begin = time.perf_counter()
_ = big_df.groupby("group")["value"].agg(["sum","mean","std","min","max"])
pd_time = time.perf_counter() - begin


begin = time.perf_counter()
_ = con.sql("""
   SELECT "group",
          SUM(worth), AVG(worth), STDDEV(worth), MIN(worth), MAX(worth)
   FROM big_df
   GROUP BY "group"
""").fetchall()
duck_time = time.perf_counter() - begin


print(f"Pandas  : {pd_time:.4f}s")
print(f"DuckDB  : {duck_time:.4f}s")
print(f"Speedup : {pd_time/duck_time:.1f}x")
print()


print("=" * 72)
print("SECTION 22: Reading Remote Parquet over HTTPS")
print("=" * 72)


strive:
   con.install_extension("httpfs")
   con.load_extension("httpfs")


   print("Querying a distant Parquet file (DuckDB taxi pattern):")
   con.sql("""
       SELECT rely(*) AS total_rows
       FROM read_parquet(
           'https://blobs.duckdb.org/information/yellow_tripdata_2010-01.parquet'
       )
   """).present()
besides Exception as e:
   print(f"(Skipped — httpfs not out there or community restricted: {e})")


print()


print("=" * 72)
print("SECTION 23: Custom Enum Types")
print("=" * 72)


con.sql("CREATE TYPE temper AS ENUM ('completely happy', 'impartial', 'unhappy')")
con.sql("""
   CREATE OR REPLACE TABLE survey(
       respondent_id INT,
       feeling       temper,
       remark       TEXT
   )
""")
con.sql("""
   INSERT INTO survey VALUES
       (1, 'completely happy',   'Great product!'),
       (2, 'impartial', 'It is okay'),
       (3, 'unhappy',     'Did not work'),
       (4, 'completely happy',   'Love it')
""")
con.sql("SELECT feeling, rely(*) AS cnt FROM survey GROUP BY feeling ORDER BY feeling").present()
print()

print("=" * 72)
print("SECTION 24: Lambda Functions in SQL")
print("=" * 72)


con.sql("""
   SELECT
       [1, 2, 3, 4, 5]                                   AS authentic,
       list_transform([1, 2, 3, 4, 5], x -> x * x)       AS squared,
       list_filter([1, 2, 3, 4, 5], x -> x % 2 = 0)     AS evens_only,
       list_reduce([1, 2, 3, 4, 5], (a, b) -> a + b)     AS whole
""").present()
print()


print("=" * 72)
print("SECTION 25: Multi-Format Export")
print("=" * 72)


summary_q = """
   SELECT
       class,
       area,
       COUNT(*)       AS orders,
       SUM(quantity)    AS income,
       AVG(quantity)    AS avg_order
   FROM gross sales
   WHERE NOT returned
   GROUP BY class, area
   ORDER BY income DESC
"""


con.sql(f"COPY ({summary_q}) TO 'abstract.csv'     (HEADER)")
con.sql(f"COPY ({summary_q}) TO 'abstract.parquet'  (FORMAT PARQUET, COMPRESSION ZSTD)")
con.sql(f"COPY ({summary_q}) TO 'abstract.json'     (FORMAT JSON, ARRAY TRUE)")


sizes = {f: os.path.getsize(f) for f in ['summary.csv','summary.parquet','summary.json']}
print("File sizes:")
for title, sz in sizes.gadgets():
   print(f"  {title:25s}  {sz:>8,} bytes")


print()


con.shut()


print("=" * 72)
print("TUTORIAL COMPLETE")
print("=" * 72)
print("""
Sections coated:
 1.  Connection administration (in-memory, persistent, config)
 2.  Synthetic information technology with generate_series()
 3.  Zero-copy querying of Pandas, Polars, PyArrow
 4.  Result conversion (.df(), .pl(), .arrow(), .fetchnumpy())
 5.  Relational API (filter / mixture / order / restrict)
 6.  Window capabilities, QUALIFY, PIVOT
 7.  Complex sorts (struct, checklist, map, unnest)
 8.  Python UDFs (scalar & vectorized)
 9.  File I/O (CSV, Parquet, JSON)
 10. Hive-partitioned Parquet writes & reads
 11. Prepared statements & parameterized queries
 12. Transactions (BEGIN / COMMIT / ROLLBACK)
 13. Appender for high-speed bulk insert
 14. Replacement scans (question Python dicts by title)
 15. SQL macros (scalar + desk macros)
 16. Recursive CTEs
 17. Full-text search (FTS extension)
 18. AsOf joins
 19. Query profiling & EXPLAIN
 20. Multi-threaded entry
 21. Performance benchmark (DuckDB vs Pandas)
 22. Remote Parquet over HTTPS (httpfs)
 23. Custom ENUM sorts
 24. Lambda capabilities in SQL
 25. Multi-format export with compression
""")