Silver to Gold: Building Business-Ready Curated Datasets

The silver layer contains clean, validated data—but it’s still millions of individual position reports. The gold layer transforms this into business-ready datasets that answer specific questions without complex SQL. Pre-aggregated data means queries that took 45 seconds now take 0.02 seconds.

Why a Gold Layer?

Layer	Data Type	Query Complexity	Users
Silver	Raw records	Complex (joins, aggregations)	Data engineers
Gold	Curated datasets	Simple (SELECT *)	Analysts, dashboards, ML

Silver layer query (complex):

SELECT timestamp, latitude, longitude, speed_over_ground
FROM silver.ais
WHERE dt = '2024-01-15' AND mmsi = 219024335
ORDER BY timestamp;
-- 500,000 rows scanned, 0.8 seconds

Gold layer query (simple):

SELECT * FROM gold.vessel_tracks
WHERE dt = '2024-01-15' AND mmsi = 219024335;
-- 1 row, 0.02 seconds (40x faster!)

The gold layer trades storage for speed—pre-computing aggregations that would otherwise be calculated on every query.

Gold Layer Datasets

Vessel Tracks

Reconstructs vessel trajectories from individual position reports.

gold/vessel_tracks/dt=YYYY-MM-DD/
├── mmsi: BIGINT (vessel identifier)
├── start_time: TIMESTAMP
├── end_time: TIMESTAMP
├── start_lat, start_lon: DOUBLE
├── end_lat, end_lon: DOUBLE
├── total_positions: INTEGER
├── avg_speed: DOUBLE
├── max_speed: DOUBLE
├── distance_nm: DOUBLE (nautical miles)
├── track_geojson: VARCHAR (GeoJSON LineString)
└── dt: DATE

Use cases:

• Visualize vessel movements on maps
• Analyze route efficiency
• Detect unusual behavior
• Predict arrival times (ML)

Port Density

Geographic heatmap of vessel activity.

gold/port_density/dt=YYYY-MM-DD/
├── grid_lat: DOUBLE
├── grid_lon: DOUBLE
├── vessel_count: INTEGER
├── position_count: INTEGER
├── avg_speed: DOUBLE
├── cell_geom: VARCHAR (WKT polygon)
└── dt: DATE

Grid resolution: 0.1° × 0.1° (~11km × 11km at equator)

Use cases:

• Identify busy shipping lanes
• Detect port congestion
• Environmental monitoring
• Traffic management

Daily Statistics

High-level metrics for monitoring.

gold/daily_stats/dt=YYYY-MM-DD/
├── source: VARCHAR
├── total_positions: BIGINT
├── unique_vessels: INTEGER
├── avg_speed: DOUBLE
├── max_speed: DOUBLE
├── geographic_bounds: VARCHAR (WKT polygon)
└── dt: DATE

Use cases:

• Data quality monitoring
• Trend analysis
• Source comparison

Implementation: Vessel Tracks

The vessel tracks asset transforms millions of position reports into thousands of trajectories:

@asset(
    partitions_def=daily_partition,
    required_resource_keys={"duckdb_conn", "minio_client"},
    ins={"dma_silver_validation": AssetIn()},
    description="Create vessel track dataset from validated silver data"
)
def vessel_tracks(context, dma_silver_validation):
    """Aggregate position reports into vessel trajectories."""
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    silver_path = f"s3://ais-lake/silver/ais/source=dma/dt={partition_date}/*.parquet"
    gold_path = f"s3://ais-lake/gold/vessel_tracks/dt={partition_date}/"

    query = f"""
    COPY (
        WITH vessel_positions AS (
            SELECT
                mmsi,
                timestamp,
                latitude,
                longitude,
                speed_over_ground,
                ROW_NUMBER() OVER (PARTITION BY mmsi ORDER BY timestamp) AS pos_num,
                COUNT(*) OVER (PARTITION BY mmsi) AS total_positions
            FROM '{silver_path}'
            WHERE mmsi IS NOT NULL
              AND latitude IS NOT NULL
              AND longitude IS NOT NULL
        ),

        first_last AS (
            SELECT
                mmsi,
                MIN(timestamp) AS start_time,
                MAX(timestamp) AS end_time,
                FIRST(latitude ORDER BY timestamp) AS start_lat,
                FIRST(longitude ORDER BY timestamp) AS start_lon,
                LAST(latitude ORDER BY timestamp) AS end_lat,
                LAST(longitude ORDER BY timestamp) AS end_lon,
                MAX(total_positions) AS total_positions
            FROM vessel_positions
            GROUP BY mmsi
        ),

        vessel_stats AS (
            SELECT
                mmsi,
                AVG(speed_over_ground) AS avg_speed,
                MAX(speed_over_ground) AS max_speed
            FROM vessel_positions
            GROUP BY mmsi
        ),

        track_geojson AS (
            SELECT
                mmsi,
                '{"type": "LineString", "coordinates": [' ||
                STRING_AGG(
                    '[' || longitude || ', ' || latitude || ']',
                    ', ' ORDER BY timestamp
                ) || ']}' AS track_geojson
            FROM vessel_positions
            GROUP BY mmsi
        )

        SELECT
            fl.mmsi,
            fl.start_time,
            fl.end_time,
            fl.start_lat,
            fl.start_lon,
            fl.end_lat,
            fl.end_lon,
            fl.total_positions,
            vs.avg_speed,
            vs.max_speed,
            tg.track_geojson,
            CAST('{partition_date}' AS DATE) AS dt
        FROM first_last fl
        LEFT JOIN vessel_stats vs ON fl.mmsi = vs.mmsi
        LEFT JOIN track_geojson tg ON fl.mmsi = tg.mmsi
        WHERE fl.total_positions >= 2

    ) TO '{gold_path}' (
        FORMAT PARQUET,
        COMPRESSION 'SNAPPY',
        ROW_GROUP_SIZE 10000
    )
    """

    result = duckdb_conn.execute(query)
    track_count = result.fetchone()[0] if result else 0
    context.log.info(f"Created {track_count:,} vessel tracks")

    return Output(value=gold_path, metadata={
        "track_count": MetadataValue.int(track_count),
        "partition_date": MetadataValue.text(partition_date),
    })

Key SQL Patterns

CTEs for clarity: Break complex aggregations into readable steps.

Window functions: FIRST() and LAST() with ORDER BY to get start/end positions.

String aggregation: Build GeoJSON directly in SQL with STRING_AGG().

Minimum threshold: Require at least 2 positions to form a track.

Implementation: Port Density

The density map grids the geographic area and counts activity per cell:

@asset(
    partitions_def=daily_partition,
    required_resource_keys={"duckdb_conn", "minio_client"},
    ins={"dma_silver_validation": AssetIn()},
    description="Create geographic density heatmap"
)
def port_density(context, dma_silver_validation):
    """Create density map of vessel activity."""
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    GRID_SIZE = 0.1  # ~11km at equator

    silver_path = f"s3://ais-lake/silver/ais/source=dma/dt={partition_date}/*.parquet"
    gold_path = f"s3://ais-lake/gold/port_density/dt={partition_date}/"

    query = f"""
    COPY (
        WITH gridded AS (
            SELECT
                ROUND(latitude / {GRID_SIZE}) * {GRID_SIZE} AS grid_lat,
                ROUND(longitude / {GRID_SIZE}) * {GRID_SIZE} AS grid_lon,
                mmsi,
                speed_over_ground
            FROM '{silver_path}'
            WHERE latitude IS NOT NULL AND longitude IS NOT NULL
        )

        SELECT
            grid_lat,
            grid_lon,
            COUNT(DISTINCT mmsi) AS vessel_count,
            COUNT(*) AS position_count,
            AVG(speed_over_ground) AS avg_speed,
            CAST('{partition_date}' AS DATE) AS dt
        FROM gridded
        GROUP BY grid_lat, grid_lon
        HAVING vessel_count > 0

    ) TO '{gold_path}' (
        FORMAT PARQUET,
        COMPRESSION 'SNAPPY'
    )
    """

    result = duckdb_conn.execute(query)
    cell_count = result.fetchone()[0] if result else 0

    return Output(value=gold_path, metadata={
        "cell_count": MetadataValue.int(cell_count),
        "grid_size_degrees": MetadataValue.float(GRID_SIZE),
    })

The ROUND(lat / GRID_SIZE) * GRID_SIZE pattern snaps each position to its grid cell, enabling efficient aggregation.

Implementation: Daily Statistics

Simple but essential for monitoring:

@asset(
    partitions_def=daily_partition,
    required_resource_keys={"duckdb_conn", "minio_client"},
    ins={"dma_silver_validation": AssetIn()},
    description="Daily summary statistics"
)
def daily_stats(context, dma_silver_validation):
    """Create daily summary statistics."""
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    silver_path = f"s3://ais-lake/silver/ais/source=dma/dt={partition_date}/*.parquet"
    gold_path = f"s3://ais-lake/gold/daily_stats/dt={partition_date}/"

    query = f"""
    COPY (
        SELECT
            'dma' AS source,
            COUNT(*) AS total_positions,
            COUNT(DISTINCT mmsi) AS unique_vessels,
            AVG(speed_over_ground) AS avg_speed,
            MAX(speed_over_ground) AS max_speed,
            MIN(latitude) AS min_lat,
            MAX(latitude) AS max_lat,
            MIN(longitude) AS min_lon,
            MAX(longitude) AS max_lon,
            CAST('{partition_date}' AS DATE) AS dt
        FROM '{silver_path}'

    ) TO '{gold_path}' (FORMAT PARQUET, COMPRESSION 'SNAPPY')
    """

    duckdb_conn.execute(query)

    return Output(value=gold_path, metadata={
        "partition_date": MetadataValue.text(partition_date),
    })

Asset Dependency Graph

Your complete pipeline now shows:

dma_ais_bronze → dma_ais_silver → dma_silver_validation ─┬─→ vessel_tracks
                                                         ├─→ port_density
                                                         └─→ daily_stats

Validation gates the gold layer—bad data never reaches business users.

Performance Optimization

Partition Pruning

Always filter by dt partition:

-- Good: Scans one partition
SELECT * FROM vessel_tracks WHERE dt = '2024-01-15';

-- Bad: Scans all partitions
SELECT * FROM vessel_tracks WHERE mmsi = 219024335;

Pre-filter in CTEs

Filter early to reduce data volume:

WITH relevant_data AS (
    SELECT * FROM silver_data
    WHERE dt = '2024-01-15'
      AND latitude BETWEEN 55 AND 58  -- Only Danish waters
)
SELECT ... FROM relevant_data

Handle Large Datasets

For very large partitions that exceed memory:

-- Process by MMSI range
WHERE mmsi BETWEEN 200000000 AND 250000000

Or use DuckDB’s temp directory:

conn.execute("SET temp_directory='/data/duckdb-temp'")

Querying Gold Data

Vessel Tracks

result = conn.execute("""
    SELECT mmsi, total_positions, avg_speed, distance_nm
    FROM 's3://ais-lake/gold/vessel_tracks/dt=2024-01-15/*.parquet'
    ORDER BY total_positions DESC
    LIMIT 10
""").fetchall()

for row in result:
    print(f"MMSI {row[0]}: {row[1]} positions, {row[2]:.1f} knots avg")

Density Heatmap

import matplotlib.pyplot as plt

df = conn.execute("""
    SELECT grid_lat, grid_lon, position_count
    FROM 's3://ais-lake/gold/port_density/dt=2024-01-15/*.parquet'
""").df()

plt.hexbin(df['grid_lon'], df['grid_lat'],
           C=df['position_count'], cmap='YlOrRd')
plt.colorbar(label='Position Count')
plt.savefig('density.png')

Vessel Track on Map

import folium
import json

track = conn.execute("""
    SELECT track_geojson
    FROM vessel_tracks
    WHERE dt = '2024-01-15' AND mmsi = 219024335
""").fetchone()[0]

m = folium.Map(location=[56.0, 11.0], zoom_start=7)
folium.GeoJson(json.loads(track)).add_to(m)
m.save("track.html")

Common Aggregation Patterns

-- Per-entity aggregation
SELECT mmsi, MIN(timestamp), MAX(timestamp), COUNT(*)
FROM positions
GROUP BY mmsi;

-- Geographic aggregation
SELECT ROUND(lat/0.1)*0.1 AS grid_lat,
       ROUND(lon/0.1)*0.1 AS grid_lon,
       COUNT(*)
FROM positions
GROUP BY grid_lat, grid_lon;

-- Time-based aggregation
SELECT DATE_TRUNC('hour', timestamp) AS hour, COUNT(*)
FROM positions
GROUP BY hour;

-- Window functions for first/last values
SELECT
    FIRST(latitude ORDER BY timestamp) AS start_lat,
    LAST(latitude ORDER BY timestamp) AS end_lat
FROM positions
GROUP BY mmsi;

What You’ve Built

Your gold layer now provides:

• Vessel tracks: Pre-built trajectories with summary statistics
• Port density: Geographic heatmaps for visualization
• Daily statistics: High-level metrics for monitoring

Gold layer queries are 10-100x faster than running aggregations on silver data. Business users get simple SELECT * queries instead of complex SQL.

What’s Next

With a complete data pipeline from bronze through gold, Chapter 9 covers adding observability—Prometheus metrics, Grafana dashboards, and alerting to keep your pipeline healthy.

---

Building analytics pipelines for your organization? We help companies design and implement production data infrastructure. Get in touch to discuss your project.