Extending Your Pipeline: New Data Sources, ML Integration, and Advanced Analytics

You’ve built a production AIS data pipeline that ingests, transforms, validates, and curates maritime vessel data. But real-world requirements evolve. This chapter shows you how to extend the pipeline for new data sources, ML integration, and advanced analytics.

Adding New Data Sources

The pipeline architecture makes adding new sources straightforward. Here’s the pattern.

Example: Adding a New AIS Provider

1. Create Downloader

# src/ingestion/downloaders/new_source_downloader.py
from minio import Minio
from datetime import date
import requests
from tenacity import retry, stop_after_attempt, wait_exponential

class NewSourceDownloader:
    """Download AIS data from new provider."""

    BASE_URL = "https://api.newsource.com/ais"

    def __init__(self, minio_client: Minio, api_key: str, bucket: str = "ais-lake"):
        self.minio_client = minio_client
        self.api_key = api_key
        self.bucket = bucket

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    def download_day(self, target_date: date, skip_if_exists: bool = True) -> str:
        object_path = f"bronze/newsource/ais-{target_date.isoformat()}.json"

        if skip_if_exists:
            try:
                self.minio_client.stat_object(self.bucket, object_path)
                return object_path
            except:
                pass

        response = requests.get(
            f"{self.BASE_URL}/data",
            params={"date": target_date.isoformat()},
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=300
        )
        response.raise_for_status()

        import io
        self.minio_client.put_object(
            bucket_name=self.bucket,
            object_name=object_path,
            data=io.BytesIO(response.content),
            length=len(response.content),
        )

        return object_path

2. Create Schema Mapping

# src/ingestion/schemas.py
NEWSOURCE_SCHEMA_MAPPING = {
    "time": "timestamp",
    "vessel_id": "mmsi",
    "lat": "latitude",
    "lon": "longitude",
    "speed": "speed_over_ground",
    "course": "course_over_ground",
}

def map_newsource_to_silver(record: dict) -> dict:
    """Map new source format to standard silver schema."""
    return {
        "timestamp": datetime.fromisoformat(record["time"]),
        "mmsi": int(record["vessel_id"]),
        "latitude": float(record["lat"]),
        "longitude": float(record["lon"]),
        "speed_over_ground": float(record.get("speed", 0)),
        "course_over_ground": float(record.get("course", 0)),
        "source": "newsource",
    }

3. Create Bronze Asset

@asset(
    partitions_def=daily_partition,
    required_resource_keys={"minio_client"},
    group_name="bronze",
)
def newsource_ais_bronze(context: AssetExecutionContext):
    """Download new source AIS data to bronze layer."""
    partition_date = context.partition_key
    api_key = os.getenv("NEWSOURCE_API_KEY")

    downloader = NewSourceDownloader(
        minio_client=context.resources.minio_client,
        api_key=api_key
    )

    object_path = downloader.download_day(
        target_date=datetime.strptime(partition_date, "%Y-%m-%d").date()
    )

    return Output(value=object_path, metadata={...})

4. Create Silver Asset

@asset(
    partitions_def=daily_partition,
    ins={"newsource_ais_bronze": AssetIn()},
    required_resource_keys={"duckdb_conn"},
    group_name="silver",
)
def newsource_ais_silver(context, newsource_ais_bronze):
    """Convert new source JSON to Parquet."""
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    bronze_path = f"s3://ais-lake/bronze/newsource/ais-{partition_date}.json"
    silver_path = f"s3://ais-lake/silver/ais/source=newsource/dt={partition_date}/"

    query = f"""
    COPY (
        SELECT
            time::TIMESTAMP AS timestamp,
            vessel_id::BIGINT AS mmsi,
            lat::DOUBLE AS latitude,
            lon::DOUBLE AS longitude,
            speed::DOUBLE AS speed_over_ground,
            'newsource' AS source,
            '{partition_date}'::DATE AS dt
        FROM read_json_auto('{bronze_path}')
        WHERE vessel_id IS NOT NULL
    ) TO '{silver_path}'
    (FORMAT PARQUET, COMPRESSION 'SNAPPY')
    """

    duckdb_conn.execute(query)
    # ... cleanup bronze, return metadata

5. Create Schedule

@schedule(
    cron_schedule="0 7 * * *",  # 07:00 UTC
    execution_timezone="UTC",
    default_status=DefaultScheduleStatus.RUNNING,
)
def daily_newsource_ingestion(context):
    yesterday = context.scheduled_execution_time - timedelta(days=1)
    partition_key = yesterday.strftime("%Y-%m-%d")

    return RunRequest(
        run_key=f"newsource_{partition_key}",
        partition_key=partition_key,
        asset_selection=["newsource_ais_bronze", "newsource_ais_silver"],
    )

Gold layer assets automatically include the new source since they read from silver/ais/**/*.parquet.

ML Integration

Exporting Training Data

Create assets that export features for ML workflows:

@asset(
    partitions_def=monthly_partition,
    ins={"vessel_tracks": AssetIn()},
    required_resource_keys={"duckdb_conn"},
    group_name="ml",
)
def vessel_behavior_features(context, vessel_tracks):
    """Export vessel track features for ML training."""
    partition_month = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    tracks_path = f"s3://ais-lake/gold/vessel_tracks/dt={partition_month}-*/*.parquet"
    output_path = f"s3://ais-lake/ml/vessel_features/month={partition_month}/"

    query = f"""
    COPY (
        WITH track_stats AS (
            SELECT
                mmsi,
                AVG(avg_speed) AS mean_speed,
                STDDEV(avg_speed) AS speed_variance,
                MAX(max_speed) AS max_speed,
                SUM(distance_nm) AS total_distance,
                COUNT(*) AS num_tracks,
                -- Behavior indicators
                SUM(CASE WHEN avg_speed < 2 THEN 1 ELSE 0 END)::FLOAT / COUNT(*) AS pct_stationary,
                SUM(CASE WHEN avg_speed > 15 THEN 1 ELSE 0 END)::FLOAT / COUNT(*) AS pct_fast,
                '{partition_month}' AS month
            FROM read_parquet('{tracks_path}')
            GROUP BY mmsi
        )
        SELECT * FROM track_stats
        WHERE num_tracks >= 5
    ) TO '{output_path}'
    (FORMAT PARQUET, COMPRESSION 'SNAPPY')
    """

    duckdb_conn.execute(query)
    return Output(value={"output_path": output_path}, metadata={...})

MLflow Integration

Log training metrics and models:

import mlflow

mlflow.set_tracking_uri("http://mlflow:5000")
mlflow.set_experiment("vessel-behavior-classification")

with mlflow.start_run():
    # Log parameters
    mlflow.log_param("data_month", "2024-01")
    mlflow.log_param("num_vessels", 15432)

    # Train model
    model = train_xgboost(X_train, y_train)

    # Log metrics
    mlflow.log_metric("accuracy", 0.94)
    mlflow.log_metric("f1_score", 0.92)

    # Log and register model
    mlflow.xgboost.log_model(model, "model")
    mlflow.register_model(
        f"runs:/{mlflow.active_run().info.run_id}/model",
        "VesselBehaviorClassifier"
    )

Sensor for ML Triggering

@sensor(
    asset_selection=["vessel_behavior_features"],
    minimum_interval_seconds=3600,
)
def ml_training_sensor(context):
    """Trigger ML training when new features materialize."""
    # Check for new materializations
    new_partitions = check_for_new_features(context)

    if not new_partitions:
        return

    # Trigger training pipeline
    for partition in new_partitions:
        trigger_ml_pipeline(
            feature_path=f"s3://ais-lake/ml/vessel_features/month={partition}/",
            model_name="vessel_behavior_classifier",
        )

Advanced Analytics Assets

Fishing Activity Detection

@asset(
    partitions_def=daily_partition,
    ins={"vessel_tracks": AssetIn()},
    required_resource_keys={"duckdb_conn"},
    group_name="analytics",
)
def fishing_activity(context, vessel_tracks):
    """Detect potential fishing activity.

    Heuristics:
    - Speed < 5 knots (typical trawling)
    - Duration > 2 hours
    - Not stationary (speed > 1 knot)
    """
    partition_date = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    tracks_path = f"s3://ais-lake/gold/vessel_tracks/dt={partition_date}/*.parquet"
    output_path = f"s3://ais-lake/analytics/fishing/dt={partition_date}/"

    query = f"""
    COPY (
        SELECT
            mmsi,
            start_time,
            end_time,
            start_lat,
            start_lon,
            avg_speed,
            (end_time - start_time) / 3600 AS duration_hours,
            CASE
                WHEN avg_speed BETWEEN 1 AND 5
                 AND (end_time - start_time) > 7200
                THEN 0.8
                WHEN avg_speed BETWEEN 1 AND 3
                 AND (end_time - start_time) > 14400
                THEN 0.9
                ELSE 0.0
            END AS fishing_likelihood,
            '{partition_date}' AS dt
        FROM read_parquet('{tracks_path}')
        WHERE fishing_likelihood > 0.5
    ) TO '{output_path}'
    (FORMAT PARQUET, COMPRESSION 'SNAPPY')
    """

    duckdb_conn.execute(query)
    return Output(value=output_path, metadata={...})

Port Traffic Analysis

@asset(
    partitions_def=monthly_partition,
    ins={"vessel_tracks": AssetIn()},
    required_resource_keys={"duckdb_conn"},
    group_name="analytics",
)
def port_traffic_analysis(context, vessel_tracks):
    """Analyze vessel traffic at major ports."""
    partition_month = context.partition_key
    duckdb_conn = context.resources.duckdb_conn

    # Define major ports
    ports = [
        {"name": "Port of Copenhagen", "lat": 55.7, "lon": 12.6, "radius_km": 5},
        {"name": "Port of Hamburg", "lat": 53.5, "lon": 9.9, "radius_km": 10},
    ]

    # For each port, count vessel visits and analyze patterns
    # ... implementation

External Integrations

REST API for Vessel Positions

Create a FastAPI service that queries your gold layer:

# api/vessel_positions.py
from fastapi import FastAPI, Query
import duckdb

app = FastAPI()

@app.get("/vessels/{mmsi}/latest")
def get_latest_position(mmsi: int):
    """Get latest known position for vessel."""
    conn = duckdb.connect()

    result = conn.execute(f"""
        SELECT end_lat, end_lon, end_time
        FROM read_parquet('s3://ais-lake/gold/vessel_tracks/**/*.parquet')
        WHERE mmsi = {mmsi}
        ORDER BY end_time DESC
        LIMIT 1
    """).fetchone()

    if not result:
        return {"error": "Vessel not found"}

    return {
        "mmsi": mmsi,
        "latitude": result[0],
        "longitude": result[1],
        "timestamp": result[2].isoformat(),
    }

@app.get("/vessels/search")
def search_vessels(
    min_lat: float, max_lat: float,
    min_lon: float, max_lon: float,
    date: str,
):
    """Search vessels in bounding box."""
    conn = duckdb.connect()

    result = conn.execute(f"""
        SELECT mmsi, end_lat, end_lon, end_time
        FROM read_parquet('s3://ais-lake/gold/vessel_tracks/dt={date}/*.parquet')
        WHERE end_lat BETWEEN {min_lat} AND {max_lat}
          AND end_lon BETWEEN {min_lon} AND {max_lon}
    """).fetchall()

    return [{"mmsi": r[0], "lat": r[1], "lon": r[2]} for r in result]

Deploy as Kubernetes service:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: vessel-api
  namespace: ais-pipeline
spec:
  replicas: 2
  template:
    spec:
      containers:
      - name: api
        image: localhost:32000/vessel-api:latest
        ports:
        - containerPort: 8000
---
apiVersion: v1
kind: Service
metadata:
  name: vessel-api
spec:
  type: LoadBalancer
  ports:
  - port: 80
    targetPort: 8000

New Source Checklist

When adding a new data source:

• Create downloader in src/ingestion/downloaders/
• Add schema mapping to src/ingestion/schemas.py
• Create bronze asset
• Create silver asset
• Register assets in __init__.py
• Create schedule
• Register schedule in repository
• Update validation suite for new schema
• Add monitoring metrics
• Test end-to-end with sample data
• Deploy to production

Follow this checklist and you’ll have a new source integrated in 2-4 hours.

What You’ve Built

Your pipeline is now a platform you can build on:

• New data sources: Pattern for adding any provider
• ML integration: Feature export and model training
• Custom analytics: Fishing detection, port traffic
• External APIs: REST endpoints for downstream consumers

Series Summary

Over these 13 chapters, you’ve built a complete production data platform:

1. Architecture: Medallion pattern on Kubernetes
2. Infrastructure: MinIO, Dagster, Prometheus, Grafana
3. Ingestion: Automated downloads with retry logic
4. Transformation: DuckDB CSV to Parquet conversion
5. Validation: Great Expectations data quality gates
6. Aggregation: Gold layer for fast analytics
7. Monitoring: Custom metrics and dashboards
8. Deployment: Docker images and Helm charts
9. Automation: Daily schedules and backfills
10. Operations: Production maintenance playbook
11. Extensions: New sources and ML workflows

The skills transfer directly—the same patterns work at any scale, from a single server to enterprise deployments.

---

Ready to build your own data platform? We help organizations design and implement production data infrastructure tailored to their needs. Get in touch to discuss your project.