Containerized Data Pipelines: Docker und Kubernetes für Platform Engineers

graph TD
    subgraph CI/CD
        GH[GitHub Actions] --> REG[Container Registry
ECR / GCR / ACR]
    end

    subgraph Kubernetes Cluster
        REG --> AIRFLOW[Apache Airflow
Kubernetes Executor]
        REG --> SPARK[Spark Operator]
        REG --> DBT[dbt Runner
CronJob]

        subgraph Namespaces
            AIRFLOW --> NS1[namespace: airflow]
            SPARK --> NS2[namespace: spark-jobs]
            DBT --> NS3[namespace: dbt]
        end

        subgraph Storage
            NS2 --> PVC[PersistentVolumeClaim
EFS / Azure Files]
            NS1 --> S3[S3 / GCS
remote logs]
        end
    end

    subgraph Monitoring
        NS1 & NS2 & NS3 --> PROM[Prometheus]
        PROM --> GRAF[Grafana]
    end

# Stage 1: Dependencies bauen
FROM python:3.11-slim AS builder

WORKDIR /build

# Build-Tools
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# Stage 2: Runtime-Image
FROM openjdk:17-slim AS runtime

# Python installieren
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3.11 \
    python3-pip \
    && rm -rf /var/lib/apt/lists/*

# Installierte Packages aus Builder kopieren
COPY --from=builder /root/.local /root/.local

# Spark installieren
ENV SPARK_VERSION=3.5.1
ENV HADOOP_VERSION=3
RUN pip install pyspark==${SPARK_VERSION}

# Pipeline-Code
WORKDIR /app
COPY src/ ./src/
COPY entrypoint.sh .

RUN chmod +x entrypoint.sh

# Non-Root-User für Security
RUN useradd -m -u 1000 spark
USER spark

ENTRYPOINT ["./entrypoint.sh"]

# Tag mit Git-SHA
IMAGE_TAG=$(git rev-parse --short HEAD)
docker build -t ecr.amazonaws.com/harbinger/pipeline:${IMAGE_TAG} .
docker push ecr.amazonaws.com/harbinger/pipeline:${IMAGE_TAG}

# Auch mit Semver-Tag für Release-Images
docker tag ecr.amazonaws.com/harbinger/pipeline:${IMAGE_TAG} \
           ecr.amazonaws.com/harbinger/pipeline:v1.4.2

helm repo add apache-airflow https://airflow.apache.org
helm repo update

helm install airflow apache-airflow/airflow \
  --namespace airflow \
  --create-namespace \
  --set executor=KubernetesExecutor \
  --set images.airflow.repository=ecr.amazonaws.com/harbinger/airflow \
  --set images.airflow.tag=2.9.1 \
  --set config.core.dags_folder=/opt/airflow/dags \
  --values airflow-values.yaml

executor: KubernetesExecutor

workers:
  resources:
    requests:
      memory: "1Gi"
      cpu: "500m"
    limits:
      memory: "4Gi"
      cpu: "2000m"

scheduler:
  resources:
    requests:
      memory: "512Mi"
      cpu: "250m"

webserver:
  replicas: 2

logs:
  persistence:
    enabled: false  # Remote-Logging nutzen
  remote:
    remoteLogging: true
    remoteBaseLogFolder: s3://harbinger-airflow-logs/

config:
  kubernetes:
    namespace: airflow
    worker_container_repository: ecr.amazonaws.com/harbinger/airflow
    worker_container_tag: 2.9.1
    delete_worker_pods: true
    delete_worker_pods_on_failure: false  # Fürs Debugging behalten

extraEnvFrom:
  - secretRef:
      name: airflow-connections

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from kubernetes.client import models as k8s
from datetime import datetime

with DAG(
    "harbinger_geopolitical_etl",
    start_date=datetime(2024, 1, 1),
    schedule="@hourly",
    catchup=False,
) as dag:

    # Schwerer Spark-Job — mehr Resources
    spark_transform = KubernetesPodOperator(
        task_id="spark_transform",
        name="spark-transform",
        namespace="airflow",
        image="ecr.amazonaws.com/harbinger/spark-pipeline:v1.4.2",
        image_pull_policy="Always",
        env_vars={
            "INPUT_PATH": "s3://harbinger-raw/events/",
            "OUTPUT_PATH": "s3://harbinger-processed/events/",
        },
        container_resources=k8s.V1ResourceRequirements(
            requests={"memory": "4Gi", "cpu": "2"},
            limits={"memory": "8Gi", "cpu": "4"},
        ),
        node_selector={"node-type": "compute-optimised"},
        tolerations=[
            k8s.V1Toleration(key="dedicated", value="spark", effect="NoSchedule")
        ],
        is_delete_operator_pod=True,
        get_logs=True,
    )

    # Leichter API-Call — minimale Resources
    api_ingest = KubernetesPodOperator(
        task_id="api_ingest",
        name="api-ingest",
        namespace="airflow",
        image="ecr.amazonaws.com/harbinger/ingest:v1.4.2",
        container_resources=k8s.V1ResourceRequirements(
            requests={"memory": "256Mi", "cpu": "100m"},
            limits={"memory": "512Mi", "cpu": "500m"},
        ),
        is_delete_operator_pod=True,
    )

    api_ingest >> spark_transform

helm repo add spark-operator https://kubeflow.github.io/spark-operator
helm install spark-operator spark-operator/spark-operator \
  --namespace spark-operator \
  --create-namespace \
  --set webhook.enable=true \
  --set metrics.enable=true \
  --set metrics.port=10254

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: harbinger-event-enrichment
  namespace: spark-jobs
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: ecr.amazonaws.com/harbinger/spark-pipeline:v1.4.2
  imagePullPolicy: Always
  mainApplicationFile: local:///app/src/enrich_events.py

  sparkVersion: "3.5.1"

  restartPolicy:
    type: OnFailure
    onFailureRetries: 3
    onFailureRetryInterval: 30

  driver:
    cores: 2
    memory: "2g"
    serviceAccount: spark
    labels:
      app: harbinger-spark
    annotations:
      iam.amazonaws.com/role: arn:aws:iam::123456789:role/SparkPipelineRole

  executor:
    cores: 4
    instances: 10
    memory: "8g"
    memoryOverhead: "2g"
    labels:
      app: harbinger-spark
    tolerations:
      - key: dedicated
        value: spark
        effect: NoSchedule

  sparkConf:
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension"
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    "spark.hadoop.fs.s3a.aws.credentials.provider": "com.amazonaws.auth.WebIdentityTokenCredentialsProvider"
    "spark.dynamicAllocation.enabled": "true"
    "spark.dynamicAllocation.minExecutors": "2"
    "spark.dynamicAllocation.maxExecutors": "50"
    "spark.dynamicAllocation.shuffleTracking.enabled": "true"

  volumes:
    - name: spark-local-dir
      emptyDir: {}

apiVersion: v1
kind: ResourceQuota
metadata:
  name: spark-jobs-quota
  namespace: spark-jobs
spec:
  hard:
    requests.cpu: "100"
    requests.memory: 400Gi
    limits.cpu: "200"
    limits.memory: 800Gi
    count/pods: "500"
    count/sparkoperator.k8s.io/v1beta2/sparkapplications: "20"

# EKS-managed Node-Group für Spark-Worker (EU-Region!)
resource "aws_eks_node_group" "spark" {
  cluster_name    = aws_eks_cluster.main.name
  node_group_name = "spark-workers"
  node_role_arn   = aws_iam_role.node.arn
  subnet_ids      = var.private_subnet_ids

  scaling_config {
    desired_size = 2
    min_size     = 0
    max_size     = 50
  }

  instance_types = ["r5.4xlarge"]  # Memory-optimized für Spark

  taint {
    key    = "dedicated"
    value  = "spark"
    effect = "NO_SCHEDULE"
  }

  labels = {
    "node-type" = "compute-optimised"
  }
}

# .github/workflows/pipeline-deploy.yml
name: Build and Deploy Pipeline

on:
  push:
    branches: [main]
    paths:
      - 'pipelines/**'
      - 'Dockerfile'

jobs:
  build:
    runs-on: ubuntu-latest
    outputs:
      image-tag: ${{ steps.meta.outputs.tags }}

    steps:
      - uses: actions/checkout@v4

      - name: AWS-Credentials konfigurieren
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789:role/GitHubActionsRole
          aws-region: eu-central-1

      - name: ECR-Login
        uses: aws-actions/amazon-ecr-login@v2

      - name: Build und Push
        uses: docker/build-push-action@v5
        with:
          push: true
          tags: |
            ${{ env.ECR_REGISTRY }}/harbinger/pipeline:${{ github.sha }}
            ${{ env.ECR_REGISTRY }}/harbinger/pipeline:latest

      - name: Security-Scan
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ env.ECR_REGISTRY }}/harbinger/pipeline:${{ github.sha }}
          severity: CRITICAL,HIGH
          exit-code: 1

  deploy:
    needs: build
    runs-on: ubuntu-latest
    steps:
      - name: Spark-Application updaten
        run: |
          kubectl set image sparkapplication/harbinger-event-enrichment \
            spark-kubernetes-driver=${{ needs.build.outputs.image-tag }} \
            -n spark-jobs

# ServiceMonitor für Prometheus
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: spark-metrics
  namespace: monitoring
spec:
  selector:
    matchLabels:
      app: harbinger-spark
  endpoints:
    - port: metrics
      interval: 15s
      path: /metrics
  namespaceSelector:
    matchNames:
      - spark-jobs