Spark on Kubernetes — Production-Grade Deployment

# Image Spark de base
FROM bitnami/spark:3.5

# Passer en root pour installer des packages
USER root

# Installer des dépendances Python
RUN pip install --no-cache-dir \
    boto3 \
    pyarrow \
    pandas

# Copier l'application
COPY app/ /app/

# Revenir à l'utilisateur non-root (sécurité)
USER 1001

# Définir le working directory
WORKDIR /app

# ============================================
# Spark Production Image
# ============================================
FROM bitnami/spark:3.5 AS base

USER root

# ---- Dépendances système ----
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && rm -rf /var/lib/apt/lists/*

# ---- Dépendances Python ----
COPY requirements.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements.txt

# ---- JARs pour connecteurs cloud ----
# AWS S3
RUN curl -sL https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar \
    -o /opt/bitnami/spark/jars/hadoop-aws-3.3.4.jar
RUN curl -sL https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.262/aws-java-sdk-bundle-1.12.262.jar \
    -o /opt/bitnami/spark/jars/aws-java-sdk-bundle-1.12.262.jar

# ---- Application ----
COPY app/ /app/

# ---- Sécurité : non-root ----
RUN chown -R 1001:1001 /app
USER 1001

WORKDIR /app

# ---- Healthcheck ----
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:4040/api/v1/applications || exit 1

# Python dependencies for Spark jobs
boto3>=1.28.0
pyarrow>=14.0.0
pandas>=2.0.0
requests>=2.31.0

# Commandes pour construire et pousser l'image
docker_commands = """
# Construire l'image
docker build -t my-registry/spark-app:1.0.0 -f Dockerfile.production .

# Tester localement
docker run --rm my-registry/spark-app:1.0.0 spark-submit --version

# Pousser vers un registry
docker push my-registry/spark-app:1.0.0

# Pour Minikube (utiliser le Docker daemon de Minikube)
eval $(minikube docker-env)
docker build -t spark-app:1.0.0 .
"""
print(docker_commands)

apiVersion: v1
kind: Namespace
metadata:
  name: spark
  labels:
    app: spark
    environment: development

# ServiceAccount pour Spark
apiVersion: v1
kind: ServiceAccount
metadata:
  name: spark-sa
  namespace: spark
---
# Role avec permissions minimales
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: spark-role
  namespace: spark
rules:
  # Gérer les pods (executors)
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["create", "get", "list", "watch", "delete"]
  # Logs des pods
  - apiGroups: [""]
    resources: ["pods/log"]
    verbs: ["get", "list"]
  # ConfigMaps pour Spark config
  - apiGroups: [""]
    resources: ["configmaps"]
    verbs: ["create", "get", "list", "watch", "delete"]
  # Services pour Spark UI
  - apiGroups: [""]
    resources: ["services"]
    verbs: ["create", "get", "delete"]
  # PersistentVolumeClaims
  - apiGroups: [""]
    resources: ["persistentvolumeclaims"]
    verbs: ["create", "get", "list", "delete"]
---
# Binding Role → ServiceAccount
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: spark-role-binding
  namespace: spark
subjects:
  - kind: ServiceAccount
    name: spark-sa
    namespace: spark
roleRef:
  kind: Role
  name: spark-role
  apiGroup: rbac.authorization.k8s.io

# Secret pour MinIO (ou S3)
apiVersion: v1
kind: Secret
metadata:
  name: minio-credentials
  namespace: spark
type: Opaque
stringData:
  AWS_ACCESS_KEY_ID: "minioadmin"
  AWS_SECRET_ACCESS_KEY: "minioadmin"
  S3_ENDPOINT: "http://minio.minio.svc.cluster.local:9000"

# Alternative : créer le secret via CLI
secret_command = """
kubectl create secret generic minio-credentials \
  --namespace=spark \
  --from-literal=AWS_ACCESS_KEY_ID=minioadmin \
  --from-literal=AWS_SECRET_ACCESS_KEY=minioadmin \
  --from-literal=S3_ENDPOINT=http://minio.minio.svc.cluster.local:9000
"""
print(secret_command)

# PVC pour les logs Spark
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: spark-logs-pvc
  namespace: spark
spec:
  accessModes:
    - ReadWriteMany  # Plusieurs pods peuvent écrire
  resources:
    requests:
      storage: 10Gi
  storageClassName: standard  # Adapter selon ton cluster
---
# PVC pour les checkpoints (streaming)
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: spark-checkpoints-pvc
  namespace: spark
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 20Gi
  storageClassName: standard

# Limiter les ressources du namespace Spark
apiVersion: v1
kind: ResourceQuota
metadata:
  name: spark-quota
  namespace: spark
spec:
  hard:
    requests.cpu: "20"        # Max 20 CPU demandés
    requests.memory: "40Gi"   # Max 40 Gi mémoire demandée
    limits.cpu: "40"          # Max 40 CPU limites
    limits.memory: "80Gi"     # Max 80 Gi mémoire limites
    pods: "50"                # Max 50 pods
---
# Defaults pour les pods sans spec
apiVersion: v1
kind: LimitRange
metadata:
  name: spark-limits
  namespace: spark
spec:
  limits:
    - type: Container
      default:
        cpu: "1"
        memory: "2Gi"
      defaultRequest:
        cpu: "500m"
        memory: "1Gi"
      max:
        cpu: "8"
        memory: "16Gi"

spark_submit_basic = """
# Spark-submit basique sur K8s (cluster mode)
spark-submit \
  --master k8s://https://<kubernetes-api-server>:6443 \
  --deploy-mode cluster \
  --name spark-pi \
  --conf spark.kubernetes.container.image=spark-app:1.0.0 \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-sa \
  local:///app/pi.py
"""
print(spark_submit_basic)

spark_submit_full = """
# Spark-submit complet pour production
spark-submit \
  # === Master & Mode ===
  --master k8s://https://$(kubectl config view -o jsonpath='{.clusters[0].cluster.server}') \
  --deploy-mode cluster \
  --name etl-job-$(date +%Y%m%d-%H%M%S) \
  \
  # === Image & Namespace ===
  --conf spark.kubernetes.container.image=my-registry/spark-app:1.0.0 \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-sa \
  \
  # === Driver Resources ===
  --conf spark.driver.cores=2 \
  --conf spark.driver.memory=4g \
  --conf spark.kubernetes.driver.request.cores=1 \
  --conf spark.kubernetes.driver.limit.cores=2 \
  \
  # === Executor Resources ===
  --conf spark.executor.instances=4 \
  --conf spark.executor.cores=2 \
  --conf spark.executor.memory=4g \
  --conf spark.kubernetes.executor.request.cores=1 \
  --conf spark.kubernetes.executor.limit.cores=2 \
  \
  # === Secrets (environnement) ===
  --conf spark.kubernetes.driver.secretKeyRef.AWS_ACCESS_KEY_ID=minio-credentials:AWS_ACCESS_KEY_ID \
  --conf spark.kubernetes.driver.secretKeyRef.AWS_SECRET_ACCESS_KEY=minio-credentials:AWS_SECRET_ACCESS_KEY \
  --conf spark.kubernetes.executor.secretKeyRef.AWS_ACCESS_KEY_ID=minio-credentials:AWS_ACCESS_KEY_ID \
  --conf spark.kubernetes.executor.secretKeyRef.AWS_SECRET_ACCESS_KEY=minio-credentials:AWS_SECRET_ACCESS_KEY \
  \
  # === S3/MinIO Config ===
  --conf spark.hadoop.fs.s3a.endpoint=http://minio.minio.svc.cluster.local:9000 \
  --conf spark.hadoop.fs.s3a.path.style.access=true \
  --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem \
  \
  # === Application ===
  local:///app/etl_job.py \
  --input s3a://bronze/data \
  --output s3a://silver/data
"""
print(spark_submit_full)

spark_ui_access = """
# 1. Trouver le pod Driver
kubectl get pods -n spark -l spark-role=driver

# 2. Port-forward vers le Spark UI
kubectl port-forward -n spark <driver-pod-name> 4040:4040

# 3. Ouvrir dans le navigateur
# http://localhost:4040

# Alternative : créer un Service
kubectl expose pod <driver-pod-name> -n spark \
  --port=4040 --target-port=4040 \
  --name=spark-ui --type=NodePort
"""
print(spark_ui_access)

operator_install = """
# Ajouter le repo Helm
helm repo add spark-operator https://kubeflow.github.io/spark-operator
helm repo update

# Installer l'opérateur
helm install spark-operator spark-operator/spark-operator \
  --namespace spark-operator \
  --create-namespace \
  --set webhook.enable=true \
  --set sparkJobNamespace=spark

# Vérifier
kubectl get pods -n spark-operator
"""
print(operator_install)

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: etl-job
  namespace: spark
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: my-registry/spark-app:1.0.0
  imagePullPolicy: Always
  mainApplicationFile: local:///app/etl_job.py
  arguments:
    - "--input"
    - "s3a://bronze/data"
    - "--output"
    - "s3a://silver/data"
  sparkVersion: "3.5.0"
  
  # Configuration Spark
  sparkConf:
    "spark.hadoop.fs.s3a.endpoint": "http://minio.minio.svc.cluster.local:9000"
    "spark.hadoop.fs.s3a.path.style.access": "true"
    "spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem"
  
  # Restart policy
  restartPolicy:
    type: OnFailure
    onFailureRetries: 3
    onFailureRetryInterval: 30
    onSubmissionFailureRetries: 3
  
  # Driver config
  driver:
    cores: 1
    coreLimit: "1200m"
    memory: "2g"
    serviceAccount: spark-sa
    labels:
      app: spark-etl
      role: driver
    envSecretKeyRefs:
      AWS_ACCESS_KEY_ID:
        name: minio-credentials
        key: AWS_ACCESS_KEY_ID
      AWS_SECRET_ACCESS_KEY:
        name: minio-credentials
        key: AWS_SECRET_ACCESS_KEY
  
  # Executor config
  executor:
    cores: 2
    coreLimit: "2400m"
    instances: 4
    memory: "4g"
    labels:
      app: spark-etl
      role: executor
    envSecretKeyRefs:
      AWS_ACCESS_KEY_ID:
        name: minio-credentials
        key: AWS_ACCESS_KEY_ID
      AWS_SECRET_ACCESS_KEY:
        name: minio-credentials
        key: AWS_SECRET_ACCESS_KEY

apiVersion: sparkoperator.k8s.io/v1beta2
kind: ScheduledSparkApplication
metadata:
  name: daily-etl
  namespace: spark
spec:
  schedule: "0 2 * * *"  # Tous les jours à 2h du matin
  concurrencyPolicy: Forbid  # Ne pas lancer si le précédent tourne encore
  successfulRunHistoryLimit: 5
  failedRunHistoryLimit: 3
  
  template:
    type: Python
    pythonVersion: "3"
    mode: cluster
    image: my-registry/spark-app:1.0.0
    mainApplicationFile: local:///app/daily_etl.py
    sparkVersion: "3.5.0"
    
    restartPolicy:
      type: OnFailure
      onFailureRetries: 2
    
    driver:
      cores: 1
      memory: "2g"
      serviceAccount: spark-sa
    
    executor:
      cores: 2
      instances: 3
      memory: "4g"

operator_commands = """
# Appliquer une SparkApplication
kubectl apply -f spark-application.yaml

# Voir le statut
kubectl get sparkapplication -n spark
kubectl describe sparkapplication etl-job -n spark

# Voir les logs du driver
kubectl logs -n spark -l spark-role=driver -f

# Supprimer
kubectl delete sparkapplication etl-job -n spark
"""
print(operator_commands)

dynamic_allocation_config = """
# Dans sparkConf ou spark-submit
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.maxExecutors=10
spark.dynamicAllocation.initialExecutors=2
spark.dynamicAllocation.executorIdleTimeout=60s
spark.dynamicAllocation.schedulerBacklogTimeout=1s

# Shuffle tracking (requis pour K8s)
spark.dynamicAllocation.shuffleTracking.enabled=true
spark.dynamicAllocation.shuffleTracking.timeout=600s
"""
print(dynamic_allocation_config)

# KEDA ScaledObject pour Spark Streaming basé sur Kafka lag
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: spark-streaming-scaler
  namespace: spark
spec:
  scaleTargetRef:
    name: spark-streaming  # Deployment à scaler
  minReplicaCount: 1
  maxReplicaCount: 10
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: kafka.kafka.svc.cluster.local:9092
        consumerGroup: spark-streaming-group
        topic: events
        lagThreshold: "100"  # Scale si lag > 100

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: gpu-job
  namespace: spark
spec:
  type: Python
  mode: cluster
  image: my-registry/spark-gpu:1.0.0
  mainApplicationFile: local:///app/gpu_job.py
  sparkVersion: "3.5.0"
  
  driver:
    cores: 1
    memory: "2g"
    serviceAccount: spark-sa
  
  executor:
    cores: 4
    instances: 2
    memory: "8g"
    # Placer les executors sur des nodes avec GPU
    affinity:
      nodeAffinity:
        requiredDuringSchedulingIgnoredDuringExecution:
          nodeSelectorTerms:
            - matchExpressions:
                - key: node-type
                  operator: In
                  values:
                    - gpu
    # Tolérer les taints GPU
    tolerations:
      - key: "nvidia.com/gpu"
        operator: "Exists"
        effect: "NoSchedule"

# Service pour Spark UI
apiVersion: v1
kind: Service
metadata:
  name: spark-ui
  namespace: spark
spec:
  selector:
    spark-role: driver
  ports:
    - port: 4040
      targetPort: 4040
  type: ClusterIP
---
# Ingress pour accès externe
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: spark-ui-ingress
  namespace: spark
spec:
  rules:
    - host: spark-ui.mycompany.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: spark-ui
                port:
                  number: 4040

apiVersion: apps/v1
kind: Deployment
metadata:
  name: spark-history-server
  namespace: spark
spec:
  replicas: 1
  selector:
    matchLabels:
      app: spark-history-server
  template:
    metadata:
      labels:
        app: spark-history-server
    spec:
      containers:
        - name: history-server
          image: bitnami/spark:3.5
          command:
            - /opt/bitnami/spark/sbin/start-history-server.sh
          env:
            - name: SPARK_HISTORY_OPTS
              value: "-Dspark.history.fs.logDirectory=s3a://spark-logs/history"
          ports:
            - containerPort: 18080
          envFrom:
            - secretRef:
                name: minio-credentials
---
apiVersion: v1
kind: Service
metadata:
  name: spark-history-server
  namespace: spark
spec:
  selector:
    app: spark-history-server
  ports:
    - port: 18080
      targetPort: 18080
  type: ClusterIP

debug_commands = """
# === Debugging Spark on K8s ===

# 1. Voir les pods et leur statut
kubectl get pods -n spark -o wide

# 2. Détails d'un pod (événements)
kubectl describe pod <pod-name> -n spark

# 3. Logs du driver
kubectl logs -n spark -l spark-role=driver

# 4. Logs d'un executor spécifique
kubectl logs -n spark <executor-pod-name>

# 5. Logs en temps réel (follow)
kubectl logs -n spark -l spark-role=driver -f

# 6. Shell dans un pod (debug)
kubectl exec -it <pod-name> -n spark -- /bin/bash

# 7. Vérifier les ressources du namespace
kubectl describe resourcequota -n spark

# 8. Vérifier les events
kubectl get events -n spark --sort-by='.lastTimestamp'

# 9. SparkApplication status
kubectl get sparkapplication -n spark
kubectl describe sparkapplication <name> -n spark
"""
print(debug_commands)

shuffle_optimizations = """
# Optimisations shuffle pour K8s

# Augmenter les timeouts
spark.network.timeout=600s
spark.shuffle.io.maxRetries=10
spark.shuffle.io.retryWait=30s

# Compression
spark.shuffle.compress=true
spark.shuffle.spill.compress=true

# Buffer sizes
spark.shuffle.file.buffer=64k
spark.reducer.maxSizeInFlight=96m

# Shuffle tracking (pour Dynamic Allocation)
spark.dynamicAllocation.shuffleTracking.enabled=true
"""
print(shuffle_optimizations)

s3a_config = """
# Optimisations S3A pour Spark on K8s

# Parallélisme
spark.hadoop.fs.s3a.connection.maximum=200
spark.hadoop.fs.s3a.threads.max=64
spark.hadoop.fs.s3a.threads.core=16

# Fast upload
spark.hadoop.fs.s3a.fast.upload=true
spark.hadoop.fs.s3a.fast.upload.buffer=bytebuffer
spark.hadoop.fs.s3a.multipart.size=104857600  # 100MB

# Committer (éviter les renames S3)
spark.sql.sources.commitProtocolClass=org.apache.spark.internal.io.cloud.PathOutputCommitProtocol
spark.sql.parquet.output.committer.class=org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter
"""
print(s3a_config)

apiVersion: v1
kind: Pod
spec:
  containers:
    - name: spark-executor
      volumeMounts:
        - name: spark-local-dir
          mountPath: /tmp/spark
      resources:
        requests:
          ephemeral-storage: "10Gi"
        limits:
          ephemeral-storage: "20Gi"
  volumes:
    - name: spark-local-dir
      emptyDir:
        medium: Memory  # Utiliser RAM pour shuffle local
        sizeLimit: "4Gi"

pod_template_usage = """
# Utiliser le pod template dans spark-submit
spark-submit \
  --conf spark.kubernetes.executor.podTemplateFile=/path/to/executor-pod-template.yaml \
  ...

# Ou dans SparkApplication
spec:
  executor:
    podTemplateFile: /path/to/executor-pod-template.yaml
"""
print(pod_template_usage)

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spot-etl
  namespace: spark
spec:
  type: Python
  mode: cluster
  image: my-registry/spark-app:1.0.0
  mainApplicationFile: local:///app/etl.py
  sparkVersion: "3.5.0"
  
  # Driver sur nodes stables (on-demand)
  driver:
    cores: 1
    memory: "2g"
    serviceAccount: spark-sa
    affinity:
      nodeAffinity:
        requiredDuringSchedulingIgnoredDuringExecution:
          nodeSelectorTerms:
            - matchExpressions:
                - key: node-type
                  operator: In
                  values:
                    - on-demand
  
  # Executors sur spot instances
  executor:
    cores: 2
    instances: 5
    memory: "4g"
    affinity:
      nodeAffinity:
        preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            preference:
              matchExpressions:
                - key: node-type
                  operator: In
                  values:
                    - spot
    tolerations:
      - key: "kubernetes.azure.com/scalesetpriority"
        operator: "Equal"
        value: "spot"
        effect: "NoSchedule"

minio_deploy = """
# Déployer MinIO avec Helm
helm repo add minio https://charts.min.io/
helm repo update

helm install minio minio/minio \
  --namespace minio \
  --create-namespace \
  --set rootUser=minioadmin \
  --set rootPassword=minioadmin \
  --set mode=standalone \
  --set resources.requests.memory=512Mi \
  --set persistence.size=10Gi

# Port-forward pour accéder à la console
kubectl port-forward -n minio svc/minio-console 9001:9001

# Créer les buckets
mc alias set myminio http://localhost:9000 minioadmin minioadmin
mc mb myminio/bronze
mc mb myminio/silver

# Uploader des données de test
mc cp data/sample.csv myminio/bronze/
"""
print(minio_deploy)

"""
ETL Job : Bronze → Silver
Lit des CSV depuis MinIO, transforme, et écrit en Parquet.
"""
import argparse
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, upper, current_timestamp


def create_spark_session():
    """Créer une SparkSession configurée pour MinIO."""
    return SparkSession.builder \
        .appName("ETL Bronze to Silver") \
        .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
        .config("spark.hadoop.fs.s3a.path.style.access", "true") \
        .getOrCreate()


def extract(spark, input_path):
    """Lire les données brutes depuis le bucket Bronze."""
    print(f"📥 Reading from {input_path}")
    df = spark.read \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .csv(input_path)
    print(f"   Loaded {df.count()} rows")
    return df


def transform(df):
    """Appliquer les transformations métier."""
    print("🔄 Transforming data")
    transformed = df \
        .dropDuplicates() \
        .dropna() \
        .withColumn("processed_at", current_timestamp())
    
    # Normalisation des colonnes string
    for col_name, dtype in transformed.dtypes:
        if dtype == "string":
            transformed = transformed.withColumn(col_name, upper(col(col_name)))
    
    print(f"   Transformed to {transformed.count()} rows")
    return transformed


def load(df, output_path):
    """Écrire les données transformées dans le bucket Silver."""
    print(f"📤 Writing to {output_path}")
    df.write \
        .mode("overwrite") \
        .parquet(output_path)
    print("   Done!")


def main():
    parser = argparse.ArgumentParser(description="ETL Job")
    parser.add_argument("--input", required=True, help="Input path (s3a://...)")
    parser.add_argument("--output", required=True, help="Output path (s3a://...)")
    args = parser.parse_args()
    
    spark = create_spark_session()
    
    try:
        # ETL Pipeline
        df = extract(spark, args.input)
        transformed = transform(df)
        load(transformed, args.output)
        
        print("✅ ETL completed successfully!")
    except Exception as e:
        print(f"❌ ETL failed: {e}")
        raise
    finally:
        spark.stop()


if __name__ == "__main__":
    main()

FROM bitnami/spark:3.5

USER root

# Dépendances Python
RUN pip install --no-cache-dir pyarrow pandas

# JARs pour S3/MinIO
RUN curl -sL https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar \
    -o /opt/bitnami/spark/jars/hadoop-aws-3.3.4.jar && \
    curl -sL https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.262/aws-java-sdk-bundle-1.12.262.jar \
    -o /opt/bitnami/spark/jars/aws-java-sdk-bundle-1.12.262.jar

# Application
COPY app/ /app/
RUN chown -R 1001:1001 /app

USER 1001
WORKDIR /app

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: etl-bronze-silver
  namespace: spark
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: spark-etl:1.0.0  # Image locale (Minikube)
  imagePullPolicy: Never  # Pour Minikube
  mainApplicationFile: local:///app/etl_job.py
  arguments:
    - "--input"
    - "s3a://bronze/sample.csv"
    - "--output"
    - "s3a://silver/data"
  sparkVersion: "3.5.0"
  
  sparkConf:
    "spark.hadoop.fs.s3a.endpoint": "http://minio.minio.svc.cluster.local:9000"
    "spark.hadoop.fs.s3a.path.style.access": "true"
    "spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem"
  
  restartPolicy:
    type: OnFailure
    onFailureRetries: 2
  
  driver:
    cores: 1
    memory: "1g"
    serviceAccount: spark-sa
    envSecretKeyRefs:
      AWS_ACCESS_KEY_ID:
        name: minio-credentials
        key: AWS_ACCESS_KEY_ID
      AWS_SECRET_ACCESS_KEY:
        name: minio-credentials
        key: AWS_SECRET_ACCESS_KEY
  
  executor:
    cores: 1
    instances: 2
    memory: "1g"
    envSecretKeyRefs:
      AWS_ACCESS_KEY_ID:
        name: minio-credentials
        key: AWS_ACCESS_KEY_ID
      AWS_SECRET_ACCESS_KEY:
        name: minio-credentials
        key: AWS_SECRET_ACCESS_KEY

deployment_script = """
#!/bin/bash
set -e

echo "🚀 Déploiement du pipeline ETL Spark on K8s"

# 1. Créer le namespace et RBAC
echo "📁 Création namespace et RBAC..."
kubectl apply -f manifests/namespace.yaml
kubectl apply -f manifests/rbac.yaml
kubectl apply -f manifests/minio-secret.yaml

# 2. Build de l'image (Minikube)
echo "🐳 Build de l'image Docker..."
eval $(minikube docker-env)
docker build -t spark-etl:1.0.0 -f docker/Dockerfile .

# 3. Vérifier que MinIO est prêt
echo "⏳ Attente de MinIO..."
kubectl wait --for=condition=ready pod -l app=minio -n minio --timeout=120s

# 4. Lancer le job Spark
echo "🔥 Lancement du job Spark..."
kubectl apply -f manifests/spark-application.yaml

# 5. Suivre le job
echo "📊 Suivi du job..."
kubectl get sparkapplication -n spark -w
"""
print(deployment_script)

verify_commands = """
# Vérifier le statut du job
kubectl get sparkapplication -n spark

# Voir les logs du driver
kubectl logs -n spark -l spark-role=driver

# Vérifier les outputs dans MinIO
mc ls myminio/silver/data/

# Lire un sample des données
mc cat myminio/silver/data/part-00000.parquet | head
"""
print(verify_commands)

# Commandes de nettoyage
cleanup_commands = """
# Supprimer les ressources Spark
kubectl delete sparkapplication --all -n spark
kubectl delete namespace spark

# Supprimer Spark Operator
helm uninstall spark-operator -n spark-operator
kubectl delete namespace spark-operator

# Supprimer MinIO
helm uninstall minio -n minio
kubectl delete namespace minio

# Arrêter Minikube (optionnel)
minikube stop
"""
print(cleanup_commands)