Data Engineering Topic
DuckDB: The Analytical Database for Local Analytics
What is DuckDB?
DuckDB is an in-process SQL OLAP database designed for analytical workloads. Think of it as "SQLite for analytics" - embedded, fast, and optimized for complex queries on large datasets.
Core Philosophy
-
Embedded: No separate server process, runs in your application
-
Columnar Storage: Optimized for analytical queries
-
Zero Configuration: Just import and use
-
High Performance: Vectorized query execution
-
Standard SQL: Full SQL support with analytical extensions
Key Features & Architecture
In-Process Architecture
text
┌─────────────────────────────────────────┐ │ Your Application Process │ │ │ │ ┌─────────────────────────────────┐ │ │ │ DuckDB Engine │ │ │ │ │ │ │ │ ┌─────────┐ ┌──────────────┐ │ │ │ │ │ Parser │ │ Optimizer │ │ │ │ │ └─────────┘ └──────────────┘ │ │ │ │ │ │ │ │ ┌─────────┐ ┌──────────────┐ │ │ │ │ │ Planner │ │ Executor │ │ │ │ │ └─────────┘ └──────────────┘ │ │ │ │ │ │ │ │ ┌───────────────────────────┐ │ │ │ │ │ Columnar Storage Engine │ │ │ │ │ │ • Vectorized Execution │ │ │ │ │ │ • Parallel Processing │ │ │ │ │ └───────────────────────────┘ │ │ │ └─────────────────────────────────┘ │ │ │ │ Local Files: │ │ • Parquet, CSV, JSON │ │ • Arrow, Pandas DataFrames │ │ │ └─────────────────────────────────────────┘
Performance Characteristics
python
import duckdb
import pandas as pd
import time
# Benchmark: DuckDB vs Pandas
df = pd.DataFrame({
'id': range(1_000_000),
'value': np.random.randn(1_000_000),
'category': np.random.choice(['A', 'B', 'C', 'D'], 1_000_000)
})
# Pandas groupby
start = time.time()
pandas_result = df.groupby('category')['value'].agg(['mean', 'sum', 'count'])
print(f"Pandas: {time.time() - start:.2f}s")
# DuckDB groupby
start = time.time()
duckdb_result = duckdb.sql("""
SELECT
category,
AVG(value) as mean,
SUM(value) as sum,
COUNT(*) as count
FROM df
GROUP BY category
""").df()
print(f"DuckDB: {time.time() - start:.2f}s")
# Typically: DuckDB 5-100x faster for analytical queries
Installation & Basic Usage
python
# Installation
# pip install duckdb
# pip install duckdb pandas pyarrow
import duckdb
import pandas as pd
import numpy as np
# Basic connection (in-memory by default)
conn = duckdb.connect() # Creates an in-memory database
# Or connect to a persistent file
conn = duckdb.connect('my_database.db') # Creates/opens a .db file
# Execute SQL directly
conn.execute("CREATE TABLE users (id INTEGER, name VARCHAR, age INTEGER)")
# Insert data
conn.execute("INSERT INTO users VALUES (1, 'Alice', 30), (2, 'Bob', 25)")
# Query data
result = conn.execute("SELECT * FROM users").fetchall()
print(result) # [(1, 'Alice', 30), (2, 'Bob', 25)]
# Query to pandas DataFrame
df = conn.execute("SELECT * FROM users").df()
Working with Different Data Sources
1. Direct File Queries
python
import duckdb
# Query Parquet files directly (no load needed)
result = duckdb.sql("""
SELECT
customer_id,
SUM(amount) as total_spent,
COUNT(*) as transaction_count
FROM 'sales_2024_*.parquet' -- Wildcard pattern
WHERE transaction_date >= '2024-01-01'
GROUP BY customer_id
HAVING total_spent > 1000
ORDER BY total_spent DESC
""").df()
# Query CSV files
result = duckdb.sql("""
SELECT *
FROM 'data.csv'
-- Automatic schema inference
WHERE column1 > 100
""").df()
# Query JSON files
result = duckdb.sql("""
SELECT
json->>'customer_id' as customer_id,
json->>'amount'::DECIMAL as amount,
json->>'timestamp'::TIMESTAMP as timestamp
FROM read_json_auto('transactions.json')
""").df()
# Multiple file formats in one query
result = duckdb.sql("""
SELECT * FROM 'sales.parquet'
UNION ALL
SELECT * FROM 'sales.csv'
UNION ALL
SELECT * FROM read_json_auto('sales.json')
""").df()
2. Integration with Pandas & Polars
python
import duckdb
import pandas as pd
import polars as pl
import numpy as np
# From Pandas DataFrame
df_pandas = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'salary': [50000, 60000, 70000]
})
# Query pandas DataFrame directly
result = duckdb.sql("""
SELECT
name,
AVG(salary) as avg_salary
FROM df_pandas
GROUP BY name
""").df()
# From Polars DataFrame
df_polars = pl.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'salary': [50000, 60000, 70000]
})
# Query polars DataFrame directly (via arrow)
result = duckdb.sql("""
SELECT * FROM df_polars
WHERE salary > 55000
""").df()
# Create DuckDB table from pandas/polars
conn.execute("CREATE TABLE employees AS SELECT * FROM df_pandas")
# or
conn.execute("CREATE TABLE employees AS SELECT * FROM df_polars")
# Two-way data transfer
# Pandas → DuckDB → Polars
duckdb_result = duckdb.sql("SELECT * FROM df_pandas WHERE salary > 55000")
polars_df = pl.from_arrow(duckdb_result.arrow()) # Fast zero-copy conversion
3. Apache Arrow Integration
python
import duckdb
import pyarrow as pa
import pyarrow.dataset as ds
# Create Arrow Table
table = pa.table({
'id': [1, 2, 3],
'value': [10.5, 20.3, 30.7]
})
# Query Arrow Table directly
result = duckdb.sql("""
SELECT
id,
value * 1.1 as value_with_tax
FROM table
""").arrow() # Returns Arrow table
# Arrow Datasets (directory of Parquet files)
dataset = ds.dataset("sales_data/", format="parquet")
result = duckdb.sql("""
SELECT
EXTRACT(YEAR FROM sale_date) as year,
SUM(amount) as yearly_revenue
FROM dataset
GROUP BY 1
""").df()
# Zero-copy between DuckDB and Arrow
# Arrow → DuckDB
duckdb.register('sales_arrow', table)
# DuckDB → Arrow
arrow_result = duckdb.sql("SELECT * FROM sales_arrow").arrow()
4. Spatial Data with DuckDB
python
import duckdb
from shapely.geometry import Point, Polygon
import geopandas as gpd
# Enable spatial extension
conn.execute("INSTALL spatial")
conn.execute("LOAD spatial")
# Create spatial table
conn.execute("""
CREATE TABLE locations AS
SELECT
1 as id,
ST_Point(-74.006, 40.7128) as geometry,
'New York' as name
UNION ALL
SELECT
2 as id,
ST_Point(-118.2437, 34.0522) as geometry,
'Los Angeles' as name
""")
# Spatial queries
result = conn.execute("""
SELECT
name,
ST_AsText(geometry) as location,
ST_Distance(
geometry,
ST_Point(-122.4194, 37.7749)
) as distance_from_sf
FROM locations
WHERE ST_DWithin(
geometry,
ST_Point(-122.4194, 37.7749),
1000000 -- Within 1000km
)
""").df()
# Load from GeoPackage/Shapefile
conn.execute("""
CREATE TABLE countries AS
SELECT * FROM ST_Read('countries.gpkg')
""")
# Spatial aggregations
result = conn.execute("""
SELECT
continent,
COUNT(*) as country_count,
SUM(ST_Area(geometry)) as total_area,
ST_Union_Agg(geometry) as combined_geometry
FROM countries
GROUP BY continent
""").df()
Advanced Features
1. Window Functions
python
import duckdb
# Complex window functions
result = duckdb.sql("""
WITH sales AS (
SELECT
customer_id,
transaction_date,
amount,
product_category,
region
FROM 'sales.parquet'
WHERE transaction_date >= '2024-01-01'
),
ranked_sales AS (
SELECT
customer_id,
transaction_date,
amount,
product_category,
region,
-- Running total per customer
SUM(amount) OVER (
PARTITION BY customer_id
ORDER BY transaction_date
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
) as cumulative_amount,
-- Rank transactions within customer
ROW_NUMBER() OVER (
PARTITION BY customer_id
ORDER BY amount DESC
) as transaction_rank,
-- Moving average (7-day window)
AVG(amount) OVER (
PARTITION BY customer_id, product_category
ORDER BY transaction_date
RANGE BETWEEN INTERVAL 7 DAYS PRECEDING
AND CURRENT ROW
) as moving_avg_7day,
-- Percent of category total
amount * 100.0 / SUM(amount) OVER (
PARTITION BY product_category
) as category_percentage,
-- Compare to customer average
amount - AVG(amount) OVER (
PARTITION BY customer_id
) as diff_from_customer_avg
FROM sales
)
SELECT *
FROM ranked_sales
WHERE transaction_rank <= 5 -- Top 5 transactions per customer
""").df()
2. Time Series Analysis
python
import duckdb
import pandas as pd
# Create time series data
dates = pd.date_range('2024-01-01', '2024-03-31', freq='D')
df = pd.DataFrame({
'date': dates,
'value': np.random.randn(len(dates)).cumsum() + 100
})
result = duckdb.sql("""
WITH time_series AS (
SELECT
date,
value,
-- Date parts
EXTRACT(YEAR FROM date) as year,
EXTRACT(MONTH FROM date) as month,
EXTRACT(DAY FROM date) as day,
EXTRACT(DOW FROM date) as day_of_week,
-- Time-based calculations
value - LAG(value, 1) OVER (ORDER BY date) as daily_change,
value - LAG(value, 7) OVER (ORDER BY date) as weekly_change,
-- Moving statistics
AVG(value) OVER (
ORDER BY date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
) as weekly_moving_avg,
STDDEV(value) OVER (
ORDER BY date
ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
) as monthly_volatility,
-- Seasonal decomposition (simplified)
value - AVG(value) OVER (
PARTITION BY EXTRACT(DOW FROM date)
ORDER BY date
ROWS BETWEEN 4 PRECEDING AND 4 FOLLOWING
) as deseasonalized
FROM df
),
aggregated AS (
SELECT
year,
month,
-- Business metrics
MIN(value) as monthly_low,
MAX(value) as monthly_high,
AVG(value) as monthly_avg,
SUM(daily_change) as net_monthly_change,
-- Volatility
STDDEV(daily_change) as daily_volatility,
-- Trend
CORR(date::INTEGER, value) as trend_correlation
FROM time_series
GROUP BY year, month
)
SELECT *
FROM aggregated
ORDER BY year, month
""").df()
3. Machine Learning with DuckDB
python
import duckdb
import pandas as pd
from sklearn.datasets import make_classification
# Enable ML extensions
conn.execute("INSTALL httpfs") # For loading datasets
conn.execute("LOAD httpfs")
# Load sample data directly from URL
conn.execute("""
CREATE TABLE iris AS
SELECT * FROM read_csv_auto('https://raw.githubusercontent.com/mwaskom/seaborn-data/master/iris.csv')
""")
# Feature engineering in SQL
features = conn.execute("""
WITH prepared_data AS (
SELECT
species,
sepal_length,
sepal_width,
petal_length,
petal_width,
-- Create features
sepal_length * sepal_width as sepal_area,
petal_length * petal_width as petal_area,
sepal_length / NULLIF(sepal_width, 0) as sepal_ratio,
petal_length / NULLIF(petal_width, 0) as petal_ratio,
-- Statistical features
(sepal_length - AVG(sepal_length) OVER()) /
NULLIF(STDDEV(sepal_length) OVER(), 0) as sepal_length_z
FROM iris
),
aggregated AS (
SELECT
species,
AVG(sepal_length) as avg_sepal_length,
AVG(petal_length) as avg_petal_length,
COUNT(*) as count,
-- Percentiles
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY sepal_width)
as median_sepal_width
FROM iris
GROUP BY species
)
SELECT * FROM prepared_data
""").df()
# Train-test split in SQL
train_test = conn.execute("""
WITH numbered AS (
SELECT *,
ROW_NUMBER() OVER () as row_num
FROM iris
)
SELECT
*,
CASE
WHEN row_num % 5 = 0 THEN 'test'
ELSE 'train'
END as dataset
FROM numbered
""").df()
# Export for ML
train_data = train_test[train_test['dataset'] == 'train']
test_data = train_test[train_test['dataset'] == 'test']
# Or use DuckDB's built-in MADlib-like functions
conn.execute("""
CREATE TABLE predictions AS
SELECT
species,
sepal_length,
-- Simple linear regression prediction
(sepal_length * 0.5 + petal_length * 0.3 + 0.2) as predicted_value
FROM iris
""")
Local Analytics Use Cases
1. Exploratory Data Analysis (EDA)
python
class LocalEDA:
"""Perform EDA entirely locally with DuckDB"""
def __init__(self, data_path):
self.conn = duckdb.connect()
self.data_path = data_path
def load_data(self):
"""Load data from various sources"""
# Auto-detect format and load
self.conn.execute(f"""
CREATE TABLE raw_data AS
SELECT * FROM read_auto('{self.data_path}')
""")
# Or explicitly
if self.data_path.endswith('.parquet'):
self.conn.execute(f"""
CREATE TABLE raw_data AS
SELECT * FROM read_parquet('{self.data_path}')
""")
elif self.data_path.endswith('.csv'):
self.conn.execute(f"""
CREATE TABLE raw_data AS
SELECT * FROM read_csv_auto('{self.data_path}')
""")
def basic_statistics(self):
"""Compute comprehensive statistics"""
stats = self.conn.execute("""
WITH column_stats AS (
SELECT
column_name,
data_type,
COUNT(*) as non_null_count,
SUM(CASE WHEN column_value IS NULL THEN 1 ELSE 0 END)
as null_count,
AVG(CASE WHEN data_type IN ('INTEGER', 'DOUBLE')
THEN column_value::DOUBLE END) as mean_value,
STDDEV(CASE WHEN data_type IN ('INTEGER', 'DOUBLE')
THEN column_value::DOUBLE END) as std_value,
MIN(column_value) as min_value,
MAX(column_value) as max_value,
COUNT(DISTINCT column_value) as distinct_count
FROM (
SELECT
UNNEST(['col1', 'col2', 'col3']) as column_name,
UNNEST(['INTEGER', 'VARCHAR', 'DOUBLE']) as data_type
) cols
CROSS JOIN raw_data
GROUP BY column_name, data_type
),
distribution AS (
SELECT
column_name,
column_value,
COUNT(*) as frequency,
COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (PARTITION BY column_name)
as percentage
FROM (
SELECT
UNNEST(['col1', 'col2']) as column_name,
UNNEST([col1, col2]) as column_value
FROM raw_data
)
WHERE column_value IS NOT NULL
GROUP BY column_name, column_value
)
SELECT * FROM column_stats
ORDER BY null_count DESC
""").df()
return stats
def correlation_analysis(self):
"""Compute correlations between numeric columns"""
correlations = self.conn.execute("""
WITH numeric_data AS (
SELECT
col1::DOUBLE as x,
col2::DOUBLE as y,
col3::DOUBLE as z
FROM raw_data
WHERE col1 IS NOT NULL
AND col2 IS NOT NULL
AND col3 IS NOT NULL
),
correlations AS (
SELECT
'col1 vs col2' as pair,
CORR(x, y) as correlation,
COUNT(*) as sample_size
FROM numeric_data
UNION ALL
SELECT
'col1 vs col3' as pair,
CORR(x, z) as correlation,
COUNT(*) as sample_size
FROM numeric_data
UNION ALL
SELECT
'col2 vs col3' as pair,
CORR(y, z) as correlation,
COUNT(*) as sample_size
FROM numeric_data
)
SELECT
pair,
ROUND(correlation, 3) as correlation,
sample_size,
CASE
WHEN ABS(correlation) > 0.7 THEN 'Strong'
WHEN ABS(correlation) > 0.3 THEN 'Moderate'
ELSE 'Weak'
END as strength
FROM correlations
ORDER BY ABS(correlation) DESC
""").df()
return correlations
def anomaly_detection(self):
"""Detect anomalies using statistical methods"""
anomalies = self.conn.execute("""
WITH stats AS (
SELECT
AVG(value) as mean,
STDDEV(value) as std,
AVG(value) - 3 * STDDEV(value) as lower_bound,
AVG(value) + 3 * STDDEV(value) as upper_bound
FROM raw_data
),
anomalies AS (
SELECT
r.*,
s.mean,
s.std,
CASE
WHEN r.value < s.lower_bound THEN 'LOW_OUTLIER'
WHEN r.value > s.upper_bound THEN 'HIGH_OUTLIER'
ELSE 'NORMAL'
END as anomaly_type,
ABS(r.value - s.mean) / NULLIF(s.std, 0) as z_score
FROM raw_data r
CROSS JOIN stats s
WHERE r.value < s.lower_bound
OR r.value > s.upper_bound
)
SELECT *
FROM anomalies
ORDER BY z_score DESC
""").df()
return anomalies
def generate_report(self):
"""Generate comprehensive EDA report"""
import matplotlib.pyplot as plt
import seaborn as sns
# Get statistics
stats = self.basic_statistics()
correlations = self.correlation_analysis()
# Create visualizations
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
# Distribution plots
for i, column in enumerate(['col1', 'col2', 'col3', 'col4'][:4]):
ax = axes[i // 2, i % 2]
data = self.conn.execute(f"""
SELECT {column}
FROM raw_data
WHERE {column} IS NOT NULL
LIMIT 1000
""").df()
if pd.api.types.is_numeric_dtype(data[column]):
ax.hist(data[column], bins=30, edgecolor='black')
ax.set_title(f'Distribution of {column}')
else:
value_counts = data[column].value_counts().head(10)
ax.bar(value_counts.index, value_counts.values)
ax.set_title(f'Top 10 values in {column}')
ax.tick_params(axis='x', rotation=45)
plt.tight_layout()
plt.savefig('eda_distributions.png')
# Generate markdown report
report = f"""
# Exploratory Data Analysis Report
## Dataset Overview
- Total rows: {len(self.conn.execute("SELECT COUNT(*) FROM raw_data").fetchone()[0])}
- Total columns: {len(stats)}
## Column Statistics
{stats.to_markdown()}
## Correlations
{correlations.to_markdown()}
## Missing Values
Missing values by column:
{(stats['null_count'] / (stats['non_null_count'] + stats['null_count']) * 100).round(2).to_markdown()}
## Visualizations
"""
with open('eda_report.md', 'w') as f:
f.write(report)
Comments (0)
No comments yet. Be the first to comment!