# Namespace for Team A's Spark jobs
apiVersion: v1
kind: Namespace
metadata:
  name: spark-team-a

---
# Enforce resource limits for Team A
apiVersion: v1
kind: ResourceQuota
metadata:
  name: team-a-quota
  namespace: spark-team-a
spec:
  hard:
    requests.cpu: "40"       # Team A can request max 40 CPU cores
    requests.memory: 160Gi   # Team A can request max 160 GB RAM
    limits.cpu: "80"         # Hard ceiling
    limits.memory: 320Gi

spark-submit \
  --master k8s://https://<k8s-api-server>:6443 \
  --deploy-mode cluster \
  --name my-pyspark-job \
  --conf spark.kubernetes.namespace=spark-jobs \
  --conf spark.kubernetes.container.image=myrepo/spark:3.5.0-python \
  \
  # Driver pod resource configuration
  --conf spark.driver.cores=2 \
  --conf spark.driver.memory=4g \
  --conf spark.kubernetes.driver.podTemplateFile=driver-template.yaml \
  \
  # Service account for the driver to call K8s API
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-sa \
  \
  # Your PySpark script
  local:///opt/spark/work-dir/my_job.py

# Service account for Spark driver
apiVersion: v1
kind: ServiceAccount
metadata:
  name: spark-sa
  namespace: spark-jobs

---
# Role allowing driver to manage pods
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: spark-role
  namespace: spark-jobs
rules:
- apiGroups: [""]
  resources: [pods, services, configmaps, persistentvolumeclaims]
  verbs: [create, get, list, watch, delete, patch]

---
# Bind the role to the service account
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: spark-rolebinding
  namespace: spark-jobs
subjects:
- kind: ServiceAccount
  name: spark-sa
roleRef:
  kind: Role
  name: spark-role
  apiGroup: rbac.authorization.k8s.io

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ShuffleTrackingDemo") \
    \
    # Enable dynamic allocation
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "2") \
    .config("spark.dynamicAllocation.maxExecutors", "50") \
    \
    # K8s-specific: shuffle tracking replaces External Shuffle Service
    .config("spark.dynamicAllocation.shuffleTracking.enabled", "true") \
    \
    # How long an executor must be idle before removal
    .config("spark.dynamicAllocation.executorIdleTimeout", "120s") \
    \
    # How long before a cached-data executor is removed
    .config("spark.dynamicAllocation.cachedExecutorIdleTimeout", "300s") \
    \
    .getOrCreate()

# Spark will now scale executor pods between 2 and 50
# based on task queue length
df = spark.read.parquet("s3://bucket/data/")
df.groupBy("category").agg({"amount": "sum"}).write.parquet("s3://bucket/output/")

# Enable graceful decommission
spark = SparkSession.builder \
    .config("spark.decommission.enabled", "true") \
    \
    # Max time to wait for graceful decommission before force-removing
    .config("spark.storage.decommission.fallbackStorage.path",
            "s3://my-bucket/spark-decommission/") \
    \
    # Shuffle data migration during decommission
    .config("spark.storage.decommission.shuffleBlocks.enabled", "true") \
    .config("spark.storage.decommission.rddBlocks.enabled", "true") \
    .getOrCreate()

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: pyspark-pi
  namespace: spark-jobs
spec:
  type: Python            # Python job
  pythonVersion: "3"
  mode: cluster            # cluster mode (driver runs as pod)
  image: myrepo/spark:3.5.0-python
  imagePullPolicy: IfNotPresent
  mainApplicationFile: local:///opt/spark/examples/pi.py

  sparkVersion: "3.5.0"
  restartPolicy:
    type: OnFailure
    onFailureRetries: 3
    onFailureRetryInterval: 10
    onSubmissionFailureRetries: 5
    onSubmissionFailureRetryInterval: 20

  driver:
    cores: 1
    coreLimit: "1200m"
    memory: "512m"
    labels:
      version: 3.5.0
    serviceAccount: spark-sa

  executor:
    cores: 1
    instances: 2
    memory: "512m"
    labels:
      version: 3.5.0

# Submit the Spark job
kubectl apply -f pyspark-pi.yaml

# Check status
kubectl get sparkapplication pyspark-pi -n spark-jobs

# Describe for detailed info
kubectl describe sparkapplication pyspark-pi -n spark-jobs

# Watch logs of the driver pod
kubectl logs -f pyspark-pi-driver -n spark-jobs

# List all Spark apps
kubectl get sparkapplications -n spark-jobs

# Delete/cancel the job
kubectl delete sparkapplication pyspark-pi -n spark-jobs

apiVersion: sparkoperator.k8s.io/v1beta2
kind: ScheduledSparkApplication
metadata:
  name: daily-etl-job
  namespace: spark-jobs
spec:
  schedule: "0 2 * * *"       # Run at 2 AM every day
  concurrencyPolicy: Forbid    # Don't allow overlapping runs
  successfulRunHistoryLimit: 3  # Keep last 3 successful runs
  failedRunHistoryLimit: 1      # Keep last 1 failed run for debugging
  template:
    spec:
      type: Python
      image: myrepo/spark:3.5.0-python
      mainApplicationFile: local:///opt/spark/work-dir/daily_etl.py
      driver:
        cores: 2
        memory: "4g"
        serviceAccount: spark-sa
      executor:
        cores: 4
        instances: 10
        memory: "8g"

# Add the Spark Operator Helm chart repo
helm repo add spark-operator https://googlecloudplatform.github.io/spark-on-k8s-operator
helm repo update

# Install the operator into its own namespace
helm install spark-operator spark-operator/spark-operator \
  --namespace spark-operator \
  --create-namespace \
  --set webhook.enable=true \          # Enable pod template mutation webhook
  --set sparkJobNamespace=spark-jobs   # Watch this namespace for SparkApplications

# Verify it's running
kubectl get pods -n spark-operator
# Output: spark-operator-xxxx   Running

apiVersion: v1
kind: Pod
metadata:
  labels:
    team: data-engineering
    env: production
spec:
  # Node selector: run only on data-plane nodes
  nodeSelector:
    node-role: spark-workers

  # Tolerations: allow running on tainted spark nodes
  tolerations:
  - key: "spark-dedicated"
    operator: "Exists"
    effect: NoSchedule

  containers:
  - name: spark             # Spark's main container (name must be "spark")
    env:
    - name: AWS_REGION
      value: us-east-1
    - name: DB_PASSWORD      # Inject secret as env var
      valueFrom:
        secretKeyRef:
          name: db-credentials
          key: password
    volumeMounts:
    - name: spark-local-dir   # Fast local SSD for shuffle
      mountPath: /tmp/spark-local

  # Sidecar for log shipping
  - name: fluentd
    image: fluent/fluentd:v1.14
    volumeMounts:
    - name: spark-logs
      mountPath: /var/log/spark

  volumes:
  - name: spark-local-dir
    emptyDir: {}             # Fast ephemeral storage on node
  - name: spark-logs
    emptyDir: {}

spark-submit \
  --master k8s://https://<k8s-api>:6443 \
  --deploy-mode cluster \
  # Point to executor pod template file
  --conf spark.kubernetes.executor.podTemplateFile=executor-pod-template.yaml \
  # Point to driver pod template file
  --conf spark.kubernetes.driver.podTemplateFile=driver-pod-template.yaml \
  local:///opt/spark/work-dir/my_job.py

spec:
  initContainers:
  - name: download-config
    image: amazon/aws-cli:latest
    command: ["aws", "s3", "cp",
               "s3://my-config-bucket/app.conf",
               "/config/app.conf"]
    volumeMounts:
    - name: config-volume
      mountPath: /config

  containers:
  - name: spark
    volumeMounts:
    - name: config-volume
      mountPath: /config   # Spark can now read /config/app.conf

  volumes:
  - name: config-volume
    emptyDir: {}

spark = SparkSession.builder \
    .appName("ResourceDemo") \
    \
    # Executor memory = Spark's JVM memory (becomes K8s memory request)
    .config("spark.executor.memory", "8g") \
    \
    # Overhead = Python/off-heap memory (added on top for K8s limit)
    .config("spark.executor.memoryOverhead", "2g") \
    # K8s memory limit = executor.memory + memoryOverhead = 10g
    \
    # CPU request for executor pod
    .config("spark.executor.cores", "4") \
    \
    # Set explicit K8s CPU limit (optional — defaults to request)
    .config("spark.kubernetes.executor.limit.cores", "4") \
    \
    # Same for driver
    .config("spark.driver.memory", "4g") \
    .config("spark.driver.memoryOverhead", "1g") \
    .config("spark.driver.cores", "2") \
    .getOrCreate()

spec:
  affinity:
    nodeAffinity:
      # MUST run on a node with this label
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
        - matchExpressions:
          - key: node-type
            operator: In
            values: [memory-optimized]

      # PREFER to run on us-east-1a (but can run elsewhere if needed)
      preferredDuringSchedulingIgnoredDuringExecution:
      - weight: 1
        preference:
          matchExpressions:
          - key: topology.kubernetes.io/zone
            operator: In
            values: [us-east-1a]

# Step 1: Taint the spark-worker nodes so nothing else runs on them
kubectl taint nodes spark-node-1 spark-node-2 \
  spark-dedicated=true:NoSchedule

# Step 2: Add toleration in the Spark pod template so Spark CAN run there

spec:
  tolerations:
  - key: "spark-dedicated"
    operator: "Equal"
    value: "true"
    effect: "NoSchedule"

  # Also target spark nodes via nodeSelector
  nodeSelector:
    node-pool: spark-workers

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: spark-kafka-processor
  namespace: spark-jobs
spec:
  jobTargetRef:
    template:
      spec:
        containers:
        - name: spark-submit
          image: myrepo/spark-submitter:latest
          command: ["spark-submit", "--master", "k8s://...", "..."]

  # Scale based on Kafka consumer lag
  triggers:
  - type: kafka
    metadata:
      bootstrapServers: kafka:9092
      consumerGroup: spark-streaming-group
      topic: events
      lagThreshold: "1000"    # Trigger a new Spark run per 1000 lagged messages

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: spark-executor-hpa
  namespace: spark-jobs
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: spark-executor-deployment
  minReplicas: 2
  maxReplicas: 30
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70   # Scale when CPU > 70%

# In AWS EKS, configure a managed node group with autoscaling
# This is typically in Terraform or eksctl config

managedNodeGroups:
- name: spark-workers
  instanceType: r5.4xlarge      # 16 vCPU, 128 GB RAM — memory optimized
  minSize: 2
  maxSize: 20
  desiredCapacity: 2
  labels:
    node-pool: spark-workers
  taints:
  - key: spark-dedicated
    value: "true"
    effect: NO_SCHEDULE

# Create a PVC for storing Spark streaming checkpoint
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: spark-checkpoint-pvc
  namespace: spark-jobs
spec:
  accessModes: [ReadWriteOnce]
  storageClassName: gp3-ssd          # Fast SSD storage class
  resources:
    requests:
      storage: 100Gi

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("StreamingWithPVC").getOrCreate()

df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "events") \
    .load()

# Checkpoint to the PVC mounted at /checkpoint in the driver pod
query = df.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/checkpoint/streaming-job") \
    .outputMode("append") \
    .start("s3://my-bucket/delta-output/")

query.awaitTermination()

spec:
  containers:
  - name: spark
    env:
    - name: SPARK_LOCAL_DIRS   # Tell Spark to use this for shuffle
      value: /mnt/spark-local
    volumeMounts:
    - name: spark-local
      mountPath: /mnt/spark-local

  volumes:
  - name: spark-local
    emptyDir:
      medium: ""           # Use node's default disk
      sizeLimit: 100Gi     # Limit shuffle size per pod

# For even faster shuffle, use Memory-backed emptyDir:
# medium: Memory  → shuffle goes to RAM (very fast but limited)

spark = SparkSession.builder \
    .appName("S3Access") \
    # Use IAM Role for Service Accounts (IRSA) — no hardcoded credentials
    .config("spark.hadoop.fs.s3a.aws.credentials.provider",
            "com.amazonaws.auth.WebIdentityTokenCredentialsProvider") \
    .config("spark.hadoop.fs.s3a.impl",
            "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.path.style.access", "false") \
    .getOrCreate()

# Read directly from S3 — credentials from pod's IAM role
df = spark.read.parquet("s3a://my-data-bucket/input/sales/")
df.write.mode("overwrite").parquet("s3a://my-data-bucket/output/sales_agg/")

volumes:
- name: shared-data
  nfs:
    server: nfs-server.spark-jobs.svc.cluster.local
    path: /shared/lookup-tables

# Mount in executor pods so all executors can read lookup tables
volumeMounts:
- name: shared-data
  mountPath: /data/lookup
  readOnly: true

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: spark-minimal-role
  namespace: spark-jobs
rules:
# Only allow what Spark actually needs
- apiGroups: [""]
  resources: [pods]
  verbs: [create, delete, get, list, watch, patch]
- apiGroups: [""]
  resources: [pods/log]        # Read executor logs
  verbs: [get, list]
- apiGroups: [""]
  resources: [configmaps]     # For Spark config
  verbs: [create, get, delete]
- apiGroups: [""]
  resources: [services]       # For driver headless service
  verbs: [create, delete, get]
# Do NOT grant: nodes, namespaces, clusterroles, secrets (only if needed)

# Create secret (base64-encoded values)
kubectl create secret generic db-creds \
  --from-literal=url=jdbc:postgresql://db:5432/mydb \
  --from-literal=user=spark_user \
  --from-literal=password=supersecret \
  -n spark-jobs

# Kubernetes stores it encrypted (if KMS encryption is enabled)

spec:
  containers:
  - name: spark
    env:
    - name: DB_URL
      valueFrom:
        secretKeyRef:
          name: db-creds
          key: url
    - name: DB_PASSWORD
      valueFrom:
        secretKeyRef:
          name: db-creds
          key: password

import os
from pyspark.sql import SparkSession

# Read DB credentials injected by Kubernetes from the Secret
db_url = os.environ["DB_URL"]
db_password = os.environ["DB_PASSWORD"]

spark = SparkSession.builder.appName("SecureJob").getOrCreate()

# Use credentials to read from JDBC without hardcoding
df = spark.read.format("jdbc") \
    .option("url", db_url) \
    .option("dbtable", "orders") \
    .option("user", "spark_user") \
    .option("password", db_password) \
    .load()

df.show()

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: spark-network-policy
  namespace: spark-jobs
spec:
  podSelector:
    matchLabels:
      spark-role: executor
  policyTypes: [Ingress, Egress]
  ingress:
  # Allow incoming connections from driver only
  - from:
    - podSelector:
        matchLabels:
          spark-role: driver
  egress:
  # Allow executors to communicate with driver
  - to:
    - podSelector:
        matchLabels:
          spark-role: driver
  # Allow executors to access S3 (via VPC endpoint = within cluster range)
  - to:
    - ipBlock:
        cidr: 10.0.0.0/8   # Internal VPC CIDR only

spark = SparkSession.builder \
    .appName("MonitoredSparkJob") \
    \
    # Enable Prometheus metrics endpoint
    .config("spark.ui.prometheus.enabled", "true") \
    \
    # Expose metrics via servlet on driver port 4040
    .config("spark.metrics.conf.*.sink.prometheusServlet.class",
            "org.apache.spark.metrics.sink.PrometheusServlet") \
    .config("spark.metrics.conf.*.sink.prometheusServlet.path",
            "/metrics/prometheus") \
    \
    # Enable executor metrics (reported to driver)
    .config("spark.executor.processTreeMetrics.enabled", "true") \
    .getOrCreate()

# Tells Prometheus Operator to scrape Spark driver metrics
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: spark-driver-monitor
  namespace: spark-jobs
spec:
  selector:
    matchLabels:
      spark-role: driver
  endpoints:
  - port: spark-ui               # Port 4040
    path: /metrics/prometheus
    interval: 15s

# Get the driver pod name
kubectl get pods -n spark-jobs | grep driver
# Output: my-spark-app-driver   Running

# Forward port 4040 to your local machine
kubectl port-forward pod/my-spark-app-driver 4040:4040 -n spark-jobs

# Now open: http://localhost:4040 in your browser
# → You see the full Spark UI: Jobs, Stages, Executors, SQL, Storage

# Alternative: Expose via a Kubernetes Service
kubectl expose pod my-spark-app-driver \
  --type=ClusterIP \
  --port=4040 \
  --name=spark-ui-service \
  -n spark-jobs

# Events for the whole namespace
kubectl get events -n spark-jobs --sort-by='.lastTimestamp'

# Events for a specific pod
kubectl describe pod my-spark-app-exec-3 -n spark-jobs
# Look for:  OOMKilled → need more executor memory
#            Insufficient memory → need bigger nodes
#            ImagePullBackOff   → Docker image issue
#            Evicted            → node was under memory pressure

# Get all failed pods in namespace
kubectl get pods -n spark-jobs --field-selector=status.phase=Failed

import logging
import json
from pyspark.sql import SparkSession

# Structured JSON logging — EFK can parse and index each field
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SparkJob")

def log_metric(event, **kwargs):
    """Emit a structured log line that EFK will parse."""
    record = {"event": event, **kwargs}
    logger.info(json.dumps(record))

spark = SparkSession.builder.appName("ProductionJob").getOrCreate()

df = spark.read.parquet("s3://bucket/input/")
row_count = df.count()

# This JSON line will be indexed in Elasticsearch by Fluentd
log_metric("data_read",
           table="input",
           row_count=row_count,
           app_id=spark.sparkContext.applicationId)

result = df.groupBy("region").count()
result.write.mode("overwrite").parquet("s3://bucket/output/")

log_metric("job_complete",
           output_table="output",
           rows_written=result.count())

# Kibana query to find all runs of this job:
# event:"job_complete" AND app_id:*