DuckDB Data Lake: Building Production-Grade Analytics with Iceberg and Delta Lake

# Install DuckDB Iceberg extension
duckdb -c "INSTALL iceberg;"
duckdb -c "LOAD iceberg;"

# Or in Python
pip install duckdb
python -c "
import duckdb
con = duckdb.connect()
con.execute('INSTALL iceberg FROM community')
con.execute('LOAD iceberg')
"

-- Configure S3 credentials
CREATE SECRET (
    TYPE S3,
    KEY_ID 'YOUR_AWS_KEY',
    SECRET 'YOUR_AWS_SECRET',
    REGION 'us-east-1'
);

-- List tables in an Iceberg catalog
SHOW TABLES IN iceberg ('s3://my-bucket/catalog/');

-- Query Iceberg table directly from S3
SELECT 
    customer_id,
    COUNT(*) as order_count,
    SUM(amount) as total_spent
FROM read_iceberg('s3://my-bucket/data/orders/*.parquet')
GROUP BY 1
ORDER BY 2 DESC
LIMIT 100;

-- Query table as of a specific timestamp
SELECT * FROM iceberg_scan(
    's3://my-bucket/data/orders/',
    {
        'snapshot_timestamp': '2026-06-20 10:00:00'
    }
);

-- Query by branch
SELECT * FROM iceberg_scan(
    's3://my-bucket/data/orders/',
    {
        'branch': 'feature-branch'
    }
);

-- Query by commit ID
SELECT * FROM iceberg_scan(
    's3://my-bucket/data/orders/',
    {
        'commit_id': 'abc123def456'
    }
);

-- Add new column to existing table
ALTER TABLE iceberg_scan('s3://my-bucket/data/orders/') 
ADD COLUMN loyalty_tier VARCHAR;

-- Modify column type
ALTER TABLE iceberg_scan('s3://my-bucket/data/orders/') 
ALTER COLUMN amount TYPE DECIMAL(18,4);

-- Rename column
ALTER TABLE iceberg_scan('s3://my-bucket/data/orders/') 
RENAME COLUMN customer_id TO user_id;

-- Create new Iceberg table
CREATE TABLE iceberg_scan('s3://my-bucket/data/new_orders/') AS
SELECT 
    order_id,
    customer_id,
    amount,
    order_date,
    status
FROM read_parquet('s3://my-bucket/raw/orders/*.parquet')
WHERE order_date >= '2026-01-01';

-- Append data
INSERT INTO iceberg_scan('s3://my-bucket/data/orders/')
SELECT * FROM read_parquet('s3://my-bucket/raw/new_orders/*.parquet');

-- Merge (upsert)
MERGE INTO iceberg_scan('s3://my-bucket/data/orders/') AS target
USING read_parquet('s3://my-bucket/raw/updates/*.parquet') AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET 
    amount = source.amount,
    status = source.status
WHEN NOT MATCHED THEN INSERT VALUES 
    (source.order_id, source.customer_id, source.amount, source.order_date, source.status);

# Install Delta Lake extension
duckdb -c "INSTALL delta;"
duckdb -c "LOAD delta;"

-- Read Delta table from S3
SELECT * FROM delta_scan('s3://my-bucket/data/delta-orders/');

-- Query with time travel
SELECT * FROM delta_scan('s3://my-bucket/data/delta-orders/', {
    'version': 42  -- Query by version
});

SELECT * FROM delta_scan('s3://my-bucket/data/delta-orders/', {
    'timestamp': '2026-06-20'  -- Query by timestamp
});

-- Create Delta table
CREATE TABLE delta_scan('s3://my-bucket/data/delta-products/') AS
SELECT 
    product_id,
    name,
    category,
    price,
    stock_quantity
FROM read_parquet('s3://my-bucket/raw/products/*.parquet');

-- Append to Delta table
INSERT INTO delta_scan('s3://my-bucket/data/delta-orders/')
SELECT * FROM read_parquet('s3://my-bucket/raw/new_orders/*.parquet');

-- Merge in Delta
CALL delta_merge(
    's3://my-bucket/data/delta-orders/',
    read_parquet('s3://my-bucket/raw/updates/*.parquet'),
    'source.order_id = target.order_id',
    ['UPDATE SET amount = source.amount, status = source.status'],
    ['INSERT VALUES (source.order_id, source.customer_id, source.amount, source.order_date, source.status)']
);

-- Compact small files
CALL delta_optimize('s3://my-bucket/data/delta-orders/');

-- Vacuum old versions
CALL delta_vacuum('s3://my-bucket/data/delta-orders/', 0);

-- Z-order optimization
CALL delta_zorder('s3://my-bucket/data/delta-orders/', 'customer_id, order_date');

import duckdb
import boto3
from datetime import datetime

class DataLakePipeline:
    def __init__(self, bucket='my-company-data-lake'):
        self.bucket = bucket
        self.s3 = boto3.client('s3')
        self.con = duckdb.connect(':memory:')
        self.con.execute('INSTALL iceberg FROM community')
        self.con.execute('LOAD iceberg')
    
    def ingest_raw_data(self, source_path, target_path):
        """Ingest raw data from various sources."""
        # Read from CSV
        self.con.execute(f"""
            COPY (
                SELECT * FROM read_csv_auto('{source_path}')
            ) TO '{target_path}' (FORMAT PARQUET)
        """)
    
    def transform_curated(self, raw_table, curated_table, transformations):
        """Apply transformations and write to curated layer."""
        self.con.execute(f"""
            CREATE OR REPLACE TABLE {curated_table} AS
            SELECT {transformations}
            FROM {raw_table}
        """)
    
    def run_analytics(self, query):
        """Run analytical queries on curated data."""
        return self.con.execute(query).fetchdf()

# Usage
pipeline = DataLakePipeline()
pipeline.ingest_raw_data(
    's3://my-company-data-lake/raw/orders/*.csv',
    's3://my-company-data-lake/curated/orders/'
)
results = pipeline.run_analytics("""
    SELECT 
        DATE_TRUNC('month', order_date) as month,
        SUM(amount) as revenue,
        COUNT(*) as orders
    FROM iceberg_scan('s3://my-company-data-lake/curated/orders/')
    GROUP BY 1
    ORDER BY 1 DESC
""")

-- Upsert new records
MERGE INTO iceberg_scan('s3://my-bucket/curated/orders/') AS target
USING (
    SELECT * FROM read_parquet('s3://my-bucket/raw/new_orders/2026-06-27/*.parquet')
) AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
    amount = source.amount,
    status = source.status,
    updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN INSERT VALUES
    (source.order_id, source.customer_id, source.amount, source.order_date, source.status);

-- DuckDB automatically optimizes Parquet file sizes
-- But you can control it explicitly:
COPY (
    SELECT * FROM iceberg_scan('s3://my-bucket/raw/orders/')
) TO 's3://my-bucket/curated/orders/' (
    FILE_FORMAT PARQUET,
    COMPRESSION ZSTD,
    PER_THREAD_OUTPUT TRUE,
    MAX_PARQUET_ROW_GROUP_SIZE 1000000
);

-- Partition by date for efficient filtering
CREATE TABLE iceberg_scan('s3://my-bucket/curated/orders/') AS
SELECT * FROM read_parquet('s3://my-bucket/raw/orders/*.parquet')
PARTITION BY (order_date);

-- Query with partition pruning
SELECT * FROM iceberg_scan('s3://my-bucket/curated/orders/')
WHERE order_date >= '2026-06-01' AND order_date < '2026-07-01';
-- Only scans June partitions

-- Enable result caching
SET enable_result_cache = TRUE;
SET result_cache_size = '2GB';

-- Cache frequently accessed tables
CREATE CACHE TABLE cached_monthly_sales AS
SELECT 
    DATE_TRUNC('month', order_date) as month,
    category,
    SUM(amount) as revenue
FROM iceberg_scan('s3://my-bucket/curated/orders/')
GROUP BY 1, 2;

-- Always use Parquet for optimal compression and query performance
COPY data TO 's3://bucket/data/' (FORMAT PARQUET, COMPRESSION ZSTD);

-- Validate data before ingesting
SELECT 
    COUNT(*) as total_rows,
    COUNT(DISTINCT order_id) as unique_orders,
    COUNT(CASE WHEN amount <= 0 THEN 1 END) as invalid_amounts,
    COUNT(CASE WHEN order_date > CURRENT_DATE THEN 1 END) as future_dates
FROM read_parquet('s3://bucket/raw/orders/*.parquet');

# S3 lifecycle policy for raw data
import boto3

s3 = boto3.client('s3')
s3.put_bucket_lifecycle_configuration(
    Bucket='my-company-data-lake',
    LifecycleConfiguration={
        'Rules': [
            {
                'ID': 'archive-raw-data',
                'Filter': {'Prefix': 'raw/'},
                'Status': 'Enabled',
                'Transitions': [
                    {'Days': 90, 'StorageClass': 'GLACIER'},
                    {'Days': 365, 'StorageClass': 'DEEP_ARCHIVE'}
                ]
            }
        ]
    }
)

-- Track query execution times
SELECT 
    query,
    avg_time_ms,
    total_runs,
    p99_time_ms
FROM query_log
WHERE database = 'analytics'
GROUP BY 1, 2, 3, 4
ORDER BY avg_time_ms DESC;