adding ebs volume example

analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/ebs-driver-pod-template.yaml

@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: ny-taxi-driver
+  namespace: emr-data-team-a
+spec:
+  nodeSelector:
+    "NodeGroupType": "SparkComputeOptimized"
+    "topology.kubernetes.io/zone": "us-west-2a"
+  tolerations:
+    - key: "spark-compute-optimized"
+      operator: "Exists"
+      effect: "NoSchedule"
+
+  containers:
+    - name: spark-kubernetes-driver # Don't change this name. EMR on EKS looking for this 

analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/ebs-executor-pod-template.yaml

@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: ny-taxi-exec
+  namespace: emr-data-team-a
+spec:
+  nodeSelector:
+    "NodeGroupType": "SparkComputeOptimized"
+    "topology.kubernetes.io/zone": "us-west-2a"
+  tolerations:
+    - key: "spark-compute-optimized"
+      operator: "Exists"
+      effect: "NoSchedule"
+
+  containers:
+    - name: spark-kubernetes-executor # Don't change this name. EMR on EKS looking for 
+
+

analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/emr-eks-karpenter-ebs.yaml

@@ -0,0 +1,23 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: emr-eks-karpenter-ebs-sc
+  namespace: emr-data-team-a
+provisioner: kubernetes.io/aws-ebs
+volumeBindingMode: WaitForFirstConsumer
+parameters:
+  type: gp3
+  fsType: ext4
+---
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: spark-driver-pvc
+  namespace: emr-data-team-a
+spec:
+  storageClassName: emr-eks-karpenter-ebs-sc
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 50Gi

analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/execute_emr_eks_ebs_job.sh

@@ -0,0 +1,113 @@
+#!/bin/bash
+
+# NOTE: Make sure to set the region before running the shell script e.g., export AWS_REGION="<your-region>"
+
+read -p "Enter the AWS Region: " AWS_REGION
+read -p "Enter the EMR Virtual Cluster ID: " EMR_VIRTUAL_CLUSTER_ID
+read -p "Enter the EMR Execution Role ARN: " EMR_EXECUTION_ROLE_ARN
+read -p "Enter the CloudWatch Log Group name: " CLOUDWATCH_LOG_GROUP
+read -p "Enter the S3 Bucket for storing PySpark Scripts, Pod Templates and Input data. For e.g., s3://<bucket-name>: " S3_BUCKET
+
+#--------------------------------------------
+# DEFAULT VARIABLES CAN BE MODIFIED
+#--------------------------------------------
+JOB_NAME='taxidata-ebs'
+EMR_EKS_RELEASE_LABEL="emr-6.7.0-latest" # Spark 3.2.1
+
+SPARK_JOB_S3_PATH="${S3_BUCKET}/${EMR_VIRTUAL_CLUSTER_ID}/${JOB_NAME}"
+SCRIPTS_S3_PATH="${SPARK_JOB_S3_PATH}/scripts"
+INPUT_DATA_S3_PATH="${SPARK_JOB_S3_PATH}/input"
+OUTPUT_DATA_S3_PATH="${SPARK_JOB_S3_PATH}/output"
+
+#--------------------------------------------
+# Copy PySpark Scripts, Pod Templates and Input data to S3 bucket
+#--------------------------------------------
+aws s3 sync "./" ${SCRIPTS_S3_PATH}
+
+#--------------------------------------------
+# NOTE: This section downloads the test data from AWS Public Dataset. You can comment this section and bring your own input data required for sample PySpark test
+# https://registry.opendata.aws/nyc-tlc-trip-records-pds/
+#--------------------------------------------
+
+mkdir -p "../input"
+# Download the input data from public data set to local folders
+wget https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2022-01.parquet -O "../input/yellow_tripdata_2022-0.parquet"
+
+# Making duplicate copies to increase the size of the data.
+max=20
+for (( i=1; i <= $max; ++i ))
+do
+    cp -rf "../input/yellow_tripdata_2022-0.parquet" "../input/yellow_tripdata_2022-${i}.parquet"
+done
+
+aws s3 sync "../input" ${INPUT_DATA_S3_PATH} # Sync from local folder to S3 path
+
+rm -rf "../input" # delete local input folder
+
+#--------------------------------------------
+# Execute Spark job
+#--------------------------------------------
+
+aws emr-containers start-job-run \
+  --virtual-cluster-id $EMR_VIRTUAL_CLUSTER_ID \
+  --name $JOB_NAME \
+  --execution-role-arn $EMR_EXECUTION_ROLE_ARN \
+  --release-label $EMR_EKS_RELEASE_LABEL \
+  --job-driver '{
+    "sparkSubmitJobDriver": {
+      "entryPoint": "'"$SCRIPTS_S3_PATH"'/pyspark-taxi-trip.py",
+      "entryPointArguments": ["'"$INPUT_DATA_S3_PATH"'",
+        "'"$OUTPUT_DATA_S3_PATH"'"
+      ],
+      "sparkSubmitParameters": "--conf spark.executor.instances=6"
+    }
+  }' \
+  --configuration-overrides '{
+    "applicationConfiguration": [
+        {
+          "classification": "spark-defaults",
+          "properties": {
+            "spark.driver.cores":"1",
+            "spark.executor.cores":"1",
+            "spark.driver.memory": "10g",
+            "spark.executor.memory": "10g",
+            "spark.kubernetes.driver.podTemplateFile":"'"$SCRIPTS_S3_PATH"'/ebs-driver-pod-template.yaml",
+            "spark.kubernetes.executor.podTemplateFile":"'"$SCRIPTS_S3_PATH"'/ebs-executor-pod-template.yaml",
+            "spark.local.dir" : "/data1,/data2",
+            "spark.kubernetes.executor.podNamePrefix":"'"$JOB_NAME"'",
+            "spark.kubernetes.driver.volumes.persistentVolumeClaim.data.options.claimName": "spark-driver-pvc",
+            "spark.kubernetes.driver.volumes.persistentVolumeClaim.data.mount.readOnly": "false",
+            "spark.kubernetes.driver.volumes.persistentVolumeClaim.data.mount.path": "/data",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.claimName": "OnDemand",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.storageClass": "emr-eks-karpenter-ebs-sc",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.sizeLimit": "50Gi",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.path": "/data",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.readOnly": "false",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-spill.options.claimName": "OnDemand",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-spill.options.storageClass": "emr-eks-karpenter-ebs-sc",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-spill.options.sizeLimit": "50Gi",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-spill.mount.path": "/var/data/spill",
+            "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-spill.mount.readOnly": "false",
+            "spark.ui.prometheus.enabled":"true",
+            "spark.executor.processTreeMetrics.enabled":"true",
+            "spark.kubernetes.driver.annotation.prometheus.io/scrape":"true",
+            "spark.kubernetes.driver.annotation.prometheus.io/path":"/metrics/executors/prometheus/",
+            "spark.kubernetes.driver.annotation.prometheus.io/port":"4040",
+            "spark.kubernetes.driver.service.annotation.prometheus.io/scrape":"true",
+            "spark.kubernetes.driver.service.annotation.prometheus.io/path":"/metrics/driver/prometheus/",
+            "spark.kubernetes.driver.service.annotation.prometheus.io/port":"4040",
+            "spark.metrics.conf.*.sink.prometheusServlet.class":"org.apache.spark.metrics.sink.PrometheusServlet",
+            "spark.metrics.conf.*.sink.prometheusServlet.path":"/metrics/driver/prometheus/",
+            "spark.metrics.conf.master.sink.prometheusServlet.path":"/metrics/master/prometheus/",
+            "spark.metrics.conf.applications.sink.prometheusServlet.path":"/metrics/applications/prometheus/"
+          }
+        }
+      ],
+    "monitoringConfiguration": {
+      "persistentAppUI":"ENABLED",
+      "cloudWatchMonitoringConfiguration": {
+        "logGroupName":"'"$CLOUDWATCH_LOG_GROUP"'",
+        "logStreamNamePrefix":"'"$JOB_NAME"'"
+      }
+    }
+  }'

analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/pyspark-taxi-trip.py

@@ -0,0 +1,66 @@
+import logging
+import sys
+from datetime import datetime
+
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import *
+from pyspark.sql import functions as f
+
+# Logging configuration
+formatter = logging.Formatter('[%(asctime)s] %(levelname)s @ line %(lineno)d: %(message)s')
+handler = logging.StreamHandler(sys.stdout)
+handler.setLevel(logging.INFO)
+handler.setFormatter(formatter)
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+logger.addHandler(handler)
+
+dt_string = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
+AppName = "NewYorkTaxiData"
+
+
+def main(args):
+
+    raw_input_folder = args[1]
+    transform_output_folder = args[2]
+
+    # Create Spark Session
+    spark = SparkSession \
+        .builder \
+        .appName(AppName + "_" + str(dt_string)) \
+        .getOrCreate()
+
+    spark.sparkContext.setLogLevel("INFO")
+    logger.info("Starting spark application")
+
+    logger.info("Reading Parquet file from S3")
+    ny_taxi_df = spark.read.parquet(raw_input_folder)
+
+    # Add additional columns to the DF
+    final_ny_taxi_df = ny_taxi_df.withColumn("current_date", f.lit(datetime.now()))
+
+    logger.info("NewYork Taxi data schema preview")
+    final_ny_taxi_df.printSchema()
+
+    logger.info("Previewing New York Taxi data sample")
+    final_ny_taxi_df.show(20, truncate=False)
+
+    logger.info("Total number of records: " + str(final_ny_taxi_df.count()))
+
+    logger.info("Write New York Taxi data to S3 transform table")
+    final_ny_taxi_df.repartition(2).write.mode("overwrite").parquet(transform_output_folder)
+
+    logger.info("Ending spark application")
+    # end spark code
+    spark.stop()
+
+    return None
+
+
+if __name__ == "__main__":
+    print(len(sys.argv))
+    if len(sys.argv) != 3:
+        print("Usage: spark-etl [input-folder] [output-folder]")
+        sys.exit(0)
+
+    main(sys.argv)

website/docs/amazon-emr-on-eks/emr-eks-karpenter.md

@@ -289,14 +289,7 @@ kubectl get pods --namespace=kube-system | grep  cluster-autoscaler # Output sho
 
 ### Execute the sample PySpark Job to trigger compute optimized Karpenter provisioner
 
-1. The following script requires three input parameters in which `EMR_VIRTUAL_CLUSTER_NAME` and `EMR_JOB_EXECUTION_ROLE_ARN` values can be extracted from `terraform apply` output values.
-2. For `S3_BUCKET`, Either create a new S3 bucket or use an existing S3 bucket to store the scripts, input and output data required to run this sample job.
-
-```text
-EMR_VIRTUAL_CLUSTER_NAME=$1   # Terraform output variable is emrcontainers_virtual_cluster_name
-S3_BUCKET=$2                  # This script requires S3 bucket as input parameter e.g., s3://<bucket-name>
-EMR_JOB_EXECUTION_ROLE_ARN=$3 # Terraform output variable is emr_on_eks_role_arn
-```
+The following script requires four input parameters `virtual_cluster_id`, `job_execution_role_arn`, `cloudwatch_log_group_name` & S3 Bucket to store PySpark scripts, Pod templates and Input data. You can get these values `terraform apply` output values or by running `terraform output`. For `S3_BUCKET`, Either create a new S3 bucket or use an existing S3 bucket.
 
 :::caution
 
@@ -306,10 +299,11 @@ This shell script downloads the test data to your local machine and uploads to S
 
 ```bash
 cd data-on-eks/analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner/
-
-./execute_emr_eks_job.sh "<EMR_VIRTUAL_CLUSTER_NAME>" \
-  "s3://<ENTER-YOUR-BUCKET-NAME>" \
-  "<EMR_JOB_EXECUTION_ROLE_ARN>"
+./execute_emr_eks_job.sh
+Enter the EMR Virtual Cluster ID: 4ucrncg6z4nd19vh1lidna2b3
+Enter the EMR Execution Role ARN: arn:aws:iam::123456789102:role/emr-eks-karpenter-emr-eks-data-team-a
+Enter the CloudWatch Log Group name: /emr-on-eks-logs/emr-eks-karpenter/emr-data-team-a
+Enter the S3 Bucket for storing PySpark Scripts, Pod Templates and Input data. For e.g., s3://<bucket-name>: s3://example-bucket
 ```
 
 Karpenter may take between 1 and 2 minutes to spin up a new compute node as specified in the provisioner templates before running the Spark Jobs.
@@ -320,6 +314,28 @@ Nodes will be drained with once the job is completed
 ```bash
 kubectl get pods --namespace=emr-data-team-a -w
 ```
+### Execute the sample PySpark job that uses EBS volumes and compute optimized Karpenter provisioner
+
+This pattern uses EBS volumes for data processing and compute optimized instances. 
+
+We will create Storageclass that will be used by drivers and executors. We'll create static Persistant Volume Claim (PVC) for the driver pod but we'll use dynamically created ebs volumes for executors. 
+
+Create StorageClass and PVC using example provided
+```bash
+kubectl apply -f emr-eks-karpenter-ebs.yaml
+```
+Let's run the job
+
+```bash
+cd data-on-eks/analytics/terraform/emr-eks-karpenter/examples/karpenter-compute-provisioner-ebs/
+./execute_emr_eks_job.sh
+Enter the EMR Virtual Cluster ID: 4ucrncg6z4nd19vh1lidna2b3
+Enter the EMR Execution Role ARN: arn:aws:iam::123456789102:role/emr-eks-karpenter-emr-eks-data-team-a
+Enter the CloudWatch Log Group name: /emr-on-eks-logs/emr-eks-karpenter/emr-data-team-a
+Enter the S3 Bucket for storing PySpark Scripts, Pod Templates and Input data. For e.g., s3://<bucket-name>: s3://example-bucket
+```
+
+You'll notice the PVC `spark-driver-pvc` will be used by driver pod but Spark will create multiple ebs volumes for executors mapped to Storageclass `emr-eks-karpenter-ebs-sc`. All dynamically created ebs volumes will be deleted once the job completes
 
 ### Execute the sample PySpark Job to trigger Memory optimized Karpenter provisioner