Bronze to Silver: Transforming CSV to Parquet with DuckDB

CSV files are great for interoperability but terrible for analytics. They’re bloated, schema-less, and force you to parse every row for every query. In this chapter, we’ll transform our raw bronze data into optimized Parquet files in the silver layer using DuckDB—achieving 10x compression and 50x faster queries.

Why Parquet?

The difference between CSV and Parquet is dramatic:

Metric	CSV	Parquet (Snappy)	Improvement
File size (daily)	2.5 GB	180 MB	93% smaller
Query time	45 sec	0.8 sec	56x faster
Schema	None	Enforced	Type safety
Compression	None	Built-in	Automatic

For a 1TB disk:

• CSV would hold ~2 years of data
• Parquet holds 10+ years

Why the Performance Difference?

Columnar storage: Parquet stores data by column, not by row. A query selecting just mmsi and speed reads only those columns—not the entire file.

Statistics: Each Parquet row group stores min/max values. When filtering WHERE mmsi = 219024335, DuckDB skips row groups where the max MMSI is less than your target.

Type-aware: No string parsing. MMSI is stored as an integer, not parsed from text on every query.

Compression: Snappy compression yields 10x reduction with nearly instant decompression.

DuckDB: SQL for Analytics

DuckDB is an in-process analytical database—think SQLite for analytics. It’s perfect for single-node deployments where Spark would be overkill.

Why DuckDB over alternatives?

Tool	Best For	Limitation
DuckDB	Single-node analytics	No distribution
pandas	Small datasets	Slow on large files
Spark	Multi-node clusters	Heavy setup, JVM overhead

For our use case, DuckDB is 10x faster than pandas and uses 1/5th the memory of Spark—all while providing a familiar SQL interface.

Direct S3 Integration

DuckDB reads and writes directly to MinIO without intermediate files:

-- Install extensions
INSTALL httpfs;
INSTALL parquet;
LOAD httpfs;
LOAD parquet;

-- Configure MinIO endpoint
SET s3_endpoint='minio:9000';
SET s3_access_key_id='your-key';
SET s3_secret_access_key='your-secret';
SET s3_use_ssl=false;
SET s3_url_style='path';

-- Query S3 directly
SELECT * FROM 's3://ais-lake/bronze/dma/aisdk-2024-01-15.csv' LIMIT 10;

No local disk needed—DuckDB streams from object storage.

Conversion Strategy

Our bronze-to-silver transformation:

1. Read compressed CSV from bronze layer
2. Validate data quality (MMSI range, coordinate bounds)
3. Standardize schema (column names, types)
4. Write partitioned Parquet to silver layer
5. Delete bronze CSV to save disk space

Partitioning Strategy

We partition by date and source:

s3://ais-lake/silver/ais/
├── source=dma/
│   ├── dt=2024-01-15/
│   │   ├── part-00000.parquet
│   │   ├── part-00001.parquet
│   │   └── part-00002.parquet
│   └── dt=2024-01-16/
│       └── ...

Why partition by date? Most queries filter by date. Partition pruning lets DuckDB skip irrelevant files entirely.

Why partition by source? Different sources can be processed and validated independently.

Implementation

DuckDB Resource

First, create a Dagster resource for DuckDB with S3 configuration:

# resources/duckdb_resource.py
import duckdb
from dagster import resource

@resource(
    config_schema={
        "memory_limit": str,
        "threads": int,
        "s3_endpoint": str,
        "s3_access_key": str,
        "s3_secret_key": str,
    }
)
def duckdb_resource(context):
    """DuckDB connection with S3 access configured."""
    config = context.resource_config

    conn = duckdb.connect(database=':memory:')

    # Set resource limits
    conn.execute(f"SET memory_limit='{config['memory_limit']}'")
    conn.execute(f"SET threads={config['threads']}")

    # Install and load extensions
    conn.execute("INSTALL httpfs; INSTALL parquet")
    conn.execute("LOAD httpfs; LOAD parquet")

    # Configure S3 (MinIO)
    conn.execute(f"SET s3_endpoint='{config['s3_endpoint']}'")
    conn.execute(f"SET s3_access_key_id='{config['s3_access_key']}'")
    conn.execute(f"SET s3_secret_access_key='{config['s3_secret_key']}'")
    conn.execute("SET s3_use_ssl=false")
    conn.execute("SET s3_url_style='path'")

    return conn

Bronze to Silver Asset

The conversion asset handles the transformation:

# assets/bronze_to_silver.py
from dagster import asset, AssetIn, Output, MetadataValue
from ..partitions import daily_partition

@asset(
    partitions_def=daily_partition,
    required_resource_keys={"minio_client", "duckdb_conn"},
    ins={"dma_ais_bronze": AssetIn()},
    description="Convert DMA bronze CSV to silver Parquet"
)
def dma_ais_silver(context, dma_ais_bronze):
    """Convert CSV to partitioned Parquet with validation.

    Input: s3://ais-lake/bronze/dma/aisdk-YYYY-MM-DD.csv
    Output: s3://ais-lake/silver/ais/source=dma/dt=YYYY-MM-DD/*.parquet
    """
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn
    minio_client = context.resources.minio_client

    bronze_path = f"s3://ais-lake/bronze/dma/aisdk-{partition_date}.csv"
    silver_path = f"s3://ais-lake/silver/ais/source=dma/dt={partition_date}/"

    context.log.info(f"Converting {bronze_path} to {silver_path}")

    query = f"""
    COPY (
        SELECT
            CAST(Timestamp AS TIMESTAMP) AS timestamp,
            CAST(MMSI AS BIGINT) AS mmsi,
            CAST(Latitude AS DOUBLE) AS latitude,
            CAST(Longitude AS DOUBLE) AS longitude,
            CAST(SOG AS DOUBLE) AS speed_over_ground,
            CAST(COG AS DOUBLE) AS course_over_ground,
            CAST(Heading AS INTEGER) AS heading,
            'dma' AS source,
            CAST('{partition_date}' AS DATE) AS dt

        FROM read_csv_auto('{bronze_path}', header=true)

        WHERE
            -- Data quality filters
            Timestamp IS NOT NULL
            AND MMSI BETWEEN 200000000 AND 799999999
            AND Latitude BETWEEN -90 AND 90
            AND Longitude BETWEEN -180 AND 180
            AND Latitude != 91.0
            AND Longitude != 181.0

    ) TO '{silver_path}' (
        FORMAT PARQUET,
        COMPRESSION 'SNAPPY',
        ROW_GROUP_SIZE 100000
    )
    """

    try:
        result = duckdb_conn.execute(query)
        row_count = result.fetchone()[0] if result else 0

        context.log.info(f"Converted {row_count:,} rows")

        # Get file info
        silver_objects = list(minio_client.list_objects(
            "ais-lake",
            prefix=f"silver/ais/source=dma/dt={partition_date}/"
        ))
        total_size = sum(obj.size for obj in silver_objects)

        # Delete bronze to save disk
        if silver_objects:
            bronze_object = f"bronze/dma/aisdk-{partition_date}.csv"
            minio_client.remove_object("ais-lake", bronze_object)
            context.log.info(f"Deleted bronze: {bronze_object}")

        return Output(
            value=silver_path,
            metadata={
                "row_count": MetadataValue.int(row_count),
                "file_count": MetadataValue.int(len(silver_objects)),
                "total_size_mb": MetadataValue.float(total_size / 1024 / 1024),
            }
        )

    except Exception as e:
        context.log.error(f"Conversion failed: {e}")
        raise

Key Implementation Details

Data validation in SQL: The WHERE clause filters invalid data during conversion:

• MMSI must be in valid range (200M-799M)
• Coordinates must be within bounds
• Special “not available” markers (91.0, 181.0) are excluded

Snappy compression: Fast decompression optimized for query performance. GZIP compresses better but decompresses 10x slower.

Row group size: 100,000 rows per group balances filtering efficiency with overhead. Smaller groups enable finer-grained statistics; larger groups reduce metadata overhead.

Bronze cleanup: After successful conversion, the bronze CSV is deleted to conserve disk space. The Parquet file becomes the source of truth.

Testing the Conversion

Verify Output Structure

After running the asset, check MinIO:

ais-lake/silver/ais/source=dma/dt=2024-01-15/
├── part-00000.parquet  (8.2 MB)
├── part-00001.parquet  (8.1 MB)
└── part-00002.parquet  (7.9 MB)

Query the Parquet Files

Test with DuckDB:

import duckdb

conn = duckdb.connect(':memory:')
# ... configure S3 ...

result = conn.execute("""
    SELECT
        COUNT(*) as rows,
        COUNT(DISTINCT mmsi) as vessels,
        AVG(speed_over_ground) as avg_speed
    FROM 's3://ais-lake/silver/ais/source=dma/dt=2024-01-15/*.parquet'
""").fetchone()

print(f"Rows: {result[0]:,}")
print(f"Vessels: {result[1]:,}")
print(f"Avg speed: {result[2]:.2f} knots")

Expected output:

Rows: 2,847,623
Vessels: 12,459
Avg speed: 8.34 knots

Performance Optimization

Memory Management

For a 32GB server, allocate about half to DuckDB:

conn.execute("SET memory_limit='16GB'")
conn.execute("SET threads=8")  # Adjust to CPU cores

Leave headroom for the OS and other services.

Query Optimization

Use partition pruning:

-- Good: Only reads one partition
SELECT * FROM 'silver/ais/*/*.parquet'
WHERE dt = '2024-01-15';

-- Bad: Reads all partitions, then filters
SELECT * FROM 'silver/ais/*/*.parquet'
WHERE timestamp > '2024-01-15';

Use column projection:

-- Good: Reads only needed columns
SELECT mmsi, latitude, longitude FROM 'data.parquet';

-- Bad: Reads all columns
SELECT * FROM 'data.parquet';

Handling Large Files

For very large files that exceed memory:

1. Process in batches by time range:

   WHERE timestamp BETWEEN '2024-01-15 00:00:00' AND '2024-01-15 06:00:00'

2. Increase timeout for slow networks:

   conn.execute("SET http_timeout=600000")  # 10 minutes

Data Quality Checks

Add validation to catch issues early:

# Minimum row count check
if row_count < 100_000:
    context.log.warning(f"Low row count: {row_count:,}")

if row_count == 0:
    raise ValueError("Conversion produced 0 rows")

# Duplicate check
duplicate_count = duckdb_conn.execute(f"""
    SELECT COUNT(*) - COUNT(DISTINCT (timestamp, mmsi))
    FROM '{silver_path}/*.parquet'
""").fetchone()[0]

if duplicate_count > 0:
    context.log.warning(f"Found {duplicate_count:,} duplicates")

What You’ve Built

Your bronze-to-silver transformation now:

• Converts CSV to Parquet with 93% size reduction
• Validates data during conversion (coordinates, MMSI range)
• Partitions by date for query optimization
• Cleans up bronze files to conserve disk space
• Tracks metadata (row counts, file sizes) in Dagster

The silver layer is now your analytical source of truth—clean, validated, and optimized for queries.

What’s Next

With clean data in the silver layer, we need to ensure ongoing quality. Chapter 7 covers automated data validation with Great Expectations, catching bad data before it reaches downstream consumers.

---

Building data transformation pipelines? We help organizations implement efficient ETL processes. Get in touch to discuss your project.