Run Spark jobs on ECI

Background

Apache Spark is a widely used open-source project for data analytics, commonly applied to big data and machine learning workloads. Starting with Apache Spark version 2.3.0, you can run and manage Spark resources on Kubernetes.

The Spark operator is designed for Spark on Kubernetes and lets developers use Custom Resource Definitions (CRDs) to submit Spark jobs to a Kubernetes cluster.

• It addresses limitations in the native Spark support for Kubernetes.

• It integrates seamlessly with components in the Kubernetes ecosystem, such as storage, monitoring, and logging.

• It supports advanced Kubernetes features such as failover, auto scaling, and scheduling optimization.

Before you begin

1. Create an ACK Serverless cluster.

   Create an ACK Serverless cluster in the Container Service for Kubernetes console. For more information, see Create an ACK Serverless cluster.

   Important

   If you need to pull images from the internet or your training job requires internet access, configure a NAT Gateway.

   You can use kubectl to manage and access the ACK Serverless cluster as follows:

   • To manage the cluster from your local computer, install and configure the kubectl client. For more information, see Obtain the kubeconfig file of a cluster and use kubectl to connect to the cluster.

   • You can also use kubectl in Cloud Shell to manage the cluster. For more information, see Use kubectl to manage a Kubernetes cluster in Cloud Shell.

2. Create an OSS bucket.

   Create an Object Storage Service (OSS) bucket to store test data, results, and logs. For instructions on how to create a bucket, see Create buckets.

Install the Spark operator

1. Install the Spark operator.

   1. In the ACK console, choose Marketplace > Marketplace in the left-side navigation pane.

   2. On the App Catalog tab, find and click ack-spark-operator.

   3. In the upper-right corner, click Deploy.

   4. In the panel, select the target cluster and complete the configuration.

2. Create a ServiceAccount, Role, and RoleBinding.

   A Spark job requires a ServiceAccount with permissions to create pods. Create the required ServiceAccount, Role, and RoleBinding. The following YAML is an example. Modify the namespace for each resource as needed.

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: spark
     namespace: default
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     namespace: default
     name: spark-role
   rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["*"]
   - apiGroups: [""]
     resources: ["services"]
     verbs: ["*"]
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: spark-role-binding
     namespace: default
   subjects:
   - kind: ServiceAccount
     name: spark
     namespace: default
   roleRef:
     kind: Role
     name: spark-role
     apiGroup: rbac.authorization.k8s.io

Build a Spark job image

You need to compile your Spark job into a JAR file and then build an image by using a Dockerfile.

This example Dockerfile uses a base Spark image from Alibaba Cloud Container Service.

FROM registry.aliyuncs.com/acs/spark:ack-2.4.5-latest
RUN mkdir -p /opt/spark/jars
# To use OSS for reading data or writing event logs, add the following JARs to the image.
ADD https://repo1.maven.org/maven2/com/aliyun/odps/hadoop-fs-oss/3.3.8-public/hadoop-fs-oss-3.3.8-public.jar $SPARK_HOME/jars
ADD https://repo1.maven.org/maven2/com/aliyun/oss/aliyun-sdk-oss/3.8.1/aliyun-sdk-oss-3.8.1.jar $SPARK_HOME/jars
ADD https://repo1.maven.org/maven2/org/aspectj/aspectjweaver/1.9.5/aspectjweaver-1.9.5.jar $SPARK_HOME/jars
ADD https://repo1.maven.org/maven2/org/jdom/jdom/1.1.3/jdom-1.1.3.jar $SPARK_HOME/jars
COPY SparkExampleScala-assembly-0.1.jar /opt/spark/jars

You can also use the Alibaba Cloud base Spark image. Alibaba Cloud provides a base image for Spark 2.4.5 that is optimized for Kubernetes scenarios, such as scheduling and elasticity, to significantly improve scheduling and startup speeds. You can enable this feature by setting the Helm chart variable enableAlibabaCloudFeatureGates: true. To achieve even faster startup speeds, set enableWebhook: false.

operatorImageName: registry.aliyuncs.com/acs/spark-operator
operatorImageVersion: ack-2.4.5-latest
operatorVersion: v2.4.5-v1beta2
imagePullPolicy: IfNotPresent
rbac:
  create: true
serviceAccounts:
  spark:
    create: true
    name: spark
  sparkoperator:
    create: true
    name: ack-spark-operator
sparkJobNamespace: "default"
enableWebhook: false
enableMetrics: true
enableAlibabaCloudFeatureGates: false

Create and submit a Spark job

Create a YAML configuration file for the Spark job and deploy it.

1. Create a spark-pi.yaml file.

   Here is a typical job manifest. For more information, see the spark-on-k8s-operator documentation.

   apiVersion: "sparkoperator.k8s.io/v1beta2"
   kind: SparkApplication
   metadata:
     name: spark-pi
     namespace: default
   spec:
     type: Scala
     mode: cluster
     image: "registry.aliyuncs.com/acs/spark:ack-2.4.5-latest"
     imagePullPolicy: Always
     mainClass: org.apache.spark.examples.SparkPi
     mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
     sparkVersion: "2.4.5"
     restartPolicy:
       type: Never
     driver:
       cores: 2
       coreLimit: "2"
       memory: "3g"
       memoryOverhead: "1g"
       labels:
         version: 2.4.5
       serviceAccount: spark
       annotations:
         k8s.aliyun.com/eci-kube-proxy-enabled: 'true'
         k8s.aliyun.com/eci-auto-imc: "true"
       tolerations:
       - key: "virtual-kubelet.io/provider"
         operator: "Exists"
     executor:
       cores: 2
       instances: 1
       memory: "3g"
       memoryOverhead: "1g"
       labels:
         version: 2.4.5
       annotations:
         k8s.aliyun.com/eci-kube-proxy-enabled: 'true'
         k8s.aliyun.com/eci-auto-imc: "true"
       tolerations:
       - key: "virtual-kubelet.io/provider"
         operator: "Exists"

2. Deploy the Spark job.

   kubectl apply -f spark-pi.yaml

Configure log collection

To collect standard output logs from the Spark driver and executor, you can inject environment variables into the envVars field to enable automatic log collection. For more information, see Customize log collection for an elastic container instance.

envVars:
   aliyun_logs_test-stdout_project: test-k8s-spark
   aliyun_logs_test-stdout_machinegroup: k8s-group-app-spark
   aliyun_logs_test-stdout: stdout

Once configured, the collected Spark executor logs are available on the Raw Log tab in the Log Service console. Each log entry includes structured fields such as __source__, __tag__:__hostname__, __tag__:__path__, and __tag__:eci_id, as well as the content of the Spark application's runtime logs, such as information about the shutdown process, including the ShutdownHookManager cleaning up directories, the BlockManager stopping, and the MemoryStore being cleared.

Configure the history server

The history server lets you audit Spark jobs. You can add the sparkConf field to the SparkApplication CRD to write event logs to OSS. The history server then reads from OSS to display the job history. The following is a configuration example:

sparkConf:
   "spark.eventLog.enabled": "true"
   "spark.eventLog.dir": "oss://bigdatastore/spark-events"
   "spark.hadoop.fs.oss.impl": "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem"
   # The endpoint of the OSS bucket, for example, oss-cn-beijing.aliyuncs.com
   "spark.hadoop.fs.oss.endpoint": "oss-cn-beijing.aliyuncs.com"
   "spark.hadoop.fs.oss.accessKeySecret": ""
   "spark.hadoop.fs.oss.accessKeyId": ""
        

Alibaba Cloud also provides a Helm chart for spark-history-server. You can search for ack-spark-history-server and install it from the Marketplace > Marketplace page in the ACK console. During installation, you must configure the OSS information in the Parameters section. The following is an example:

oss:
  enableOSS: true
  # Please input your accessKeyId
  alibabaCloudAccessKeyId: ""
  # Please input your accessKeySecret
  alibabaCloudAccessKeySecret: ""
  # oss bucket endpoint such as oss-cn-beijing.aliyuncs.com
  alibabaCloudOSSEndpoint: "oss-cn-beijing.aliyuncs.com"
  # oss file path such as oss://bucket-name/path
  eventsDir: "oss://bigdatastore/spark-events"

After installation, find the external endpoint for ack-spark-history-server on the Services page and access it to view archived jobs. On the Spark History Server page, you can view a list of completed Spark applications. The list includes columns such as App id, App name, Started, Completed, Duration, Spark user, Last updated, and Event log. You can also download the corresponding event log by clicking the Download button.

Verify the job result

1. Check the status of the pods.

   kubectl get pods

   Expected output:

   NAME                            READY      STATUS     RESTARTS   AGE
   spark-pi-1547981232122-driver   1/1       Running    0          12s
   spark-pi-1547981232122-exec-1   1/1       Running    0          3s

2. View the real-time Spark UI.

   kubectl port-forward spark-pi-1547981232122-driver 4040:4040

3. Check the status of the SparkApplication.

   kubectl describe sparkapplication spark-pi

   Expected output:

   Name:         spark-pi
   Namespace:    default
   Labels:       <none>
   Annotations:  kubectl.kubernetes.io/last-applied-configuration:
                   {"apiVersion":"sparkoperator.k8s.io/v1alpha1","kind":"SparkApplication","metadata":{"annotations":{},"name":"spark-pi","namespace":"default"...}}
   API Version:  sparkoperator.k8s.io/v1alpha1
   Kind:         SparkApplication
   Metadata:
     Creation Timestamp:  2019-01-20T10:47:08Z
     Generation:          1
     Resource Version:    4923532
     Self Link:           /apis/sparkoperator.k8s.io/v1alpha1/namespaces/default/sparkapplications/spark-pi
     UID:                 bbe7445c-1ca0-11e9-9ad4-062fd7c19a7b
   Spec:
     Deps:
     Driver:
       Core Limit:  200m
       Cores:       0.1
       Labels:
         Version:        2.4.0
       Memory:           512m
       Service Account:  spark
       Volume Mounts:
         Mount Path:  /tmp
         Name:        test-volume
     Executor:
       Cores:      1
       Instances:  1
       Labels:
         Version:  2.4.0
       Memory:     512m
       Volume Mounts:
         Mount Path:         /tmp
         Name:               test-volume
     Image:                  gcr.io/spark-operator/spark:v2.4.0
     Image Pull Policy:      Always
     Main Application File:  local:///opt/spark/examples/jars/spark-examples_2.11-2.4.0.jar
     Main Class:             org.apache.spark.examples.SparkPi
     Mode:                   cluster
     Restart Policy:
       Type:  Never
     Type:    Scala
     Volumes:
       Host Path:
         Path:  /tmp
         Type:  Directory
       Name:    test-volume
   Status:
     Application State:
       Error Message:
       State:          COMPLETED
     Driver Info:
       Pod Name:             spark-pi-driver
       Web UI Port:          31182
       Web UI Service Name:  spark-pi-ui-svc
     Execution Attempts:     1
     Executor State:
       Spark - Pi - 1547981232122 - Exec - 1:  COMPLETED
     Last Submission Attempt Time:             2019-01-20T10:47:14Z
     Spark Application Id:                     spark-application-1547981285779
     Submission Attempts:                      1
     Termination Time:                         2019-01-20T10:48:56Z
   Events:
     Type    Reason                     Age                 From            Message
     ----    ------                     ----                ----            -------
     Normal  SparkApplicationAdded      55m                 spark-operator  SparkApplication spark-pi was added, Enqueuing it for submission
     Normal  SparkApplicationSubmitted  55m                 spark-operator  SparkApplication spark-pi was submitted successfully
     Normal  SparkDriverPending         55m (x2 over 55m)   spark-operator  Driver spark-pi-driver is pending
     Normal  SparkExecutorPending       54m (x3 over 54m)   spark-operator  Executor spark-pi-1547981232122-exec-1 is pending
     Normal  SparkExecutorRunning       53m (x4 over 54m)   spark-operator  Executor spark-pi-1547981232122-exec-1 is running
     Normal  SparkDriverRunning         53m (x12 over 55m)  spark-operator  Driver spark-pi-driver is running
     Normal  SparkExecutorCompleted     53m (x2 over 53m)   spark-operator  Executor spark-pi-1547981232122-exec-1 completed

4. Check the logs for the result.

   NAME                                      READY     STATUS      RESTARTS   AGE
   spark-pi-1547981232122-driver   0/1       Completed   0          1m

   When the SparkApplication or driver pod status is COMPLETED, check the logs for the result.

   kubectl logs spark-pi-1547981232122-driver
   Pi is roughly 3.152155760778804