CI/CD Pipelines für Databricks: Produktionsreifer Guide

harbinger-pipelines/
├── .github/
│   └── workflows/
│       ├── ci.yml
│       └── deploy.yml
├── databricks.yml
├── resources/
│   ├── jobs.yml
│   └── pipelines.yml
├── src/
│   ├── pipelines/
│   │   ├── bronze/
│   │   │   └── ingest_events.py
│   │   ├── silver/
│   │   │   └── clean_events.py
│   │   └── gold/
│   │       └── aggregate_signals.py
│   └── utils/
│       ├── schema_utils.py
│       └── quality_checks.py
├── tests/
│   ├── unit/
│   │   └── test_clean_events.py
│   └── integration/
│       └── test_pipeline_e2e.py
├── notebooks/
├── pyproject.toml
└── requirements.txt

bundle:
  name: harbinger-pipelines

variables:
  environment:
    description: Deployment environment
    default: dev
  catalog:
    description: Unity Catalog name
    default: harbinger_dev

targets:
  dev:
    mode: development
    default: true
    workspace:
      host: https://adb-7405613637854743.3.azuredatabricks.net
    variables:
      environment: dev
      catalog: harbinger_dev

  staging:
    mode: development
    workspace:
      host: https://adb-7405613637854743.3.azuredatabricks.net
    variables:
      environment: staging
      catalog: harbinger_staging

  prod:
    mode: production
    workspace:
      host: https://adb-7405613637854743.3.azuredatabricks.net
    variables:
      environment: prod
      catalog: harbinger_prod
    run_as:
      service_principal_name: harbinger-prod-sp

resources:
  jobs:
    events_ingestion_job:
      name: "harbinger-events-ingestion-${var.environment}"
      schedule:
        quartz_cron_expression: "0 0 * * * ?"
        timezone_id: "UTC"
      tasks:
        - task_key: ingest_bronze
          python_wheel_task:
            package_name: harbinger_pipelines
            entry_point: ingest_events
          job_cluster_key: default

        - task_key: clean_silver
          depends_on:
            - task_key: ingest_bronze
          python_wheel_task:
            package_name: harbinger_pipelines
            entry_point: clean_events
          job_cluster_key: default

      job_clusters:
        - job_cluster_key: default
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "Standard_DS3_v2"
            autoscale:
              min_workers: 2
              max_workers: 8

name: CI

on:
  pull_request:
    branches: [main, staging]

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: |
          pip install -r requirements-dev.txt
          pip install -e .

      - name: Run linting
        run: |
          ruff check src/ tests/
          mypy src/

      - name: Run unit tests
        run: pytest tests/unit/ -v --tb=short

      - name: Install Databricks CLI
        run: |
          curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

      - name: Validate bundle
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST_DEV }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN_DEV }}
        run: databricks bundle validate --target dev

      - name: Deploy to dev
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST_DEV }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN_DEV }}
        run: databricks bundle deploy --target dev

name: Deploy

on:
  push:
    branches: [main]

jobs:
  deploy-staging:
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - uses: actions/checkout@v4

      - name: Install Databricks CLI
        run: curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

      - name: Deploy to staging
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN_STAGING }}
        run: databricks bundle deploy --target staging

      - name: Run integration tests
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN_STAGING }}
        run: pytest tests/integration/ -v --timeout=300

  deploy-prod:
    runs-on: ubuntu-latest
    needs: deploy-staging
    environment: production
    steps:
      - uses: actions/checkout@v4

      - name: Install Databricks CLI
        run: curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh

      - name: Deploy to production
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN_PROD }}
        run: databricks bundle deploy --target prod

# src/pipelines/silver/clean_events.py

from pyspark.sql import DataFrame
from pyspark.sql.functions import col, trim, upper, when

def clean_event_type(df: DataFrame) -> DataFrame:
    return df.withColumn(
        "event_type",
        when(upper(trim(col("event_type"))).isin(["CONFLICT", "WAR", "BATTLE"]), "CONFLICT")
        .when(upper(trim(col("event_type"))).isin(["NATURAL_DISASTER", "EARTHQUAKE", "FLOOD"]), "DISASTER")
        .otherwise("UNKNOWN")
    )

def filter_low_quality_events(df: DataFrame, min_severity: float = 0.5) -> DataFrame:
    return df.filter(col("severity") >= min_severity)

# tests/unit/test_clean_events.py

import pytest
from pyspark.sql import SparkSession
from src.pipelines.silver.clean_events import clean_event_type, filter_low_quality_events

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[1]").appName("test").getOrCreate()

def test_clean_event_type_normalizes_aliases(spark):
    data = [("war", 5.0), ("FLOOD", 7.0), ("political", 3.0)]
    df = spark.createDataFrame(data, ["event_type", "severity"])
    result = clean_event_type(df)
    types = {row.event_type for row in result.collect()}
    assert "CONFLICT" in types
    assert "DISASTER" in types
    assert "UNKNOWN" in types

def test_filter_removes_low_severity(spark):
    data = [("CONFLICT", 0.3), ("DISASTER", 0.8), ("CONFLICT", 0.5)]
    df = spark.createDataFrame(data, ["event_type", "severity"])
    result = filter_low_quality_events(df, min_severity=0.5)
    assert result.count() == 2

main  (production)
  |
  +-- feature/add-gdelt-source     PR -> main
  +-- feature/optimize-silver      PR -> main
  +-- hotfix/fix-null-handling     PR -> main (fast-track)

gh secret set DATABRICKS_TOKEN_PROD --body "dapi..."
gh secret set DATABRICKS_TOKEN_STAGING --body "dapi..."
gh secret set DATABRICKS_HOST --body "https://adb-xxx.azuredatabricks.net"

# Check what is deployed in each target
databricks bundle summary --target prod

# Run a job manually to verify
databricks bundle run events_ingestion_job --target prod