Comet Kubernetes Support

Comet Docker Images

Run the following command from the root of this repository to build the Comet Docker image. The image is based on apache/spark:4.1.3 and adds the Comet JAR for Spark 4.1 and Scala 2.13 to $SPARK_HOME/jars.

docker build -t apache/datafusion-comet -f kube/Dockerfile .

Push the image to a registry that your cluster can pull from, and use that image name in the image field of the examples below.

Example Spark Submit

The exact syntax will vary depending on the Kubernetes distribution, but an example spark-submit command can be found here.

Helm chart

Install helm Spark operator for Kubernetes

# Add the Helm repository
helm repo add spark-operator https://kubeflow.github.io/spark-operator
helm repo update

# Install the operator into the spark-operator namespace and wait for deployments to be ready
helm install spark-operator spark-operator/spark-operator --namespace spark-operator --create-namespace --wait

Check the operator is deployed

helm status --namespace spark-operator spark-operator

NAME: my-release
NAMESPACE: spark-operator
STATUS: deployed
REVISION: 1
TEST SUITE: None

Create an example Spark SQL application file named comet-dataframe-pi.yaml. This application uses Spark’s DataFrame API, so it exercises Comet without requiring an external data source.

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: comet-dataframe-pi
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: apache/datafusion-comet
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.sql.SparkDataFramePi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.13-4.1.3.jar
  arguments:
    - "10"
  sparkConf:
    "spark.executor.extraClassPath": "/opt/spark/jars/comet-spark-spark4.1_2.13-1.3.0-SNAPSHOT.jar"
    "spark.driver.extraClassPath": "/opt/spark/jars/comet-spark-spark4.1_2.13-1.3.0-SNAPSHOT.jar"
    "spark.plugins": "org.apache.spark.CometPlugin"
    "spark.memory.offHeap.enabled": "true"
    "spark.memory.offHeap.size": "1g"
    "spark.executor.memoryOverhead": "1g"
    "spark.comet.enabled": "true"
    "spark.comet.exec.enabled": "true"
    "spark.comet.exec.range.enabled": "true"
    "spark.comet.explain.native.enabled": "true"
    "spark.comet.shuffle.enabled": "true"
    "spark.comet.shuffle.mode": "auto"
    "spark.shuffle.manager": "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager"
    "spark.kubernetes.executor.deleteOnTermination": "false"
  sparkVersion: 4.1.3
  driver:
    labels:
      version: 4.1.3
    cores: 1
    coreLimit: 1200m
    memory: 512m
    serviceAccount: spark-operator-spark
  executor:
    labels:
      version: 4.1.3
    instances: 1
    cores: 1
    coreLimit: 1200m
    memory: 512m

Refer to Comet builds

Run Apache Spark application with Comet enabled

kubectl apply -f comet-dataframe-pi.yaml
sparkapplication.sparkoperator.k8s.io/comet-dataframe-pi created

Check application status

kubectl get sparkapp comet-dataframe-pi

NAME                 STATUS      ATTEMPTS   START   FINISH   AGE
comet-dataframe-pi   COMPLETED   1          ...     ...      ...

To check more runtime details

kubectl describe sparkapplication comet-dataframe-pi

....
Events:
  Type    Reason                     Age    From                          Message
  ----    ------                     ----   ----                          -------
  Normal  SparkApplicationSubmitted  8m15s  spark-application-controller  SparkApplication comet-dataframe-pi was submitted successfully
  Normal  SparkDriverRunning         7m18s  spark-application-controller  Driver comet-dataframe-pi-driver is running
  Normal  SparkExecutorPending       7m11s  spark-application-controller  Executor [comet-dataframe-pi-68732195ab217303-exec-1] is pending
  Normal  SparkExecutorRunning       7m10s  spark-application-controller  Executor [comet-dataframe-pi-68732195ab217303-exec-1] is running
  Normal  SparkExecutorCompleted     7m5s   spark-application-controller  Executor [comet-dataframe-pi-68732195ab217303-exec-1] completed
  Normal  SparkDriverCompleted       7m4s   spark-application-controller  Driver comet-dataframe-pi-driver completed

Get Driver Logs

kubectl logs comet-dataframe-pi-driver

The driver log contains the computed approximation (Pi is roughly ...). The example retains the completed executor pod so that you can inspect its log for the native plan emitted by spark.comet.explain.native.enabled and confirm that Comet executed the DataFrame query. Delete the application with kubectl delete sparkapplication comet-dataframe-pi when you finish.

More info on Kube Spark operator