Processing Data on Kubernetes with Iceberg and Spark

1. Data Processing Architecture

It is mainly divided into four layers:

Processing capability layer: Spark on Kubernetes provides streaming data processing capability
Data management layer: Iceberg provides dataset access operations such as ACID and tables
Storage layer: Hive MetaStore manages Iceberg table metadata, PostgreSQL serves as the storage backend for Hive MetaStore, and S3 serves as the data storage backend
Resource layer: Kubernetes manages the cluster’s compute, storage, and network resources, providing a unified resource management capability to the layers above

1.1 Spark

Apache Spark is an open-source cluster computing framework, originally developed by the AMPLab at UC Berkeley. Compared with Hadoop’s MapReduce, which writes intermediate data to disk after a job completes, Spark uses in-memory computing technology and can perform analysis and computation in memory before the data has been written to disk.

The tasks it can accomplish include:

Batch processing, extraction, transformation, loading, processing, etc.
Stream processing, real-time analytics, event processing
SQL queries, integrating with data sources such as Hive and Iceberg, providing powerful query optimization
Machine learning, MLlib in Spark supports common machine learning algorithms such as regression, classification, and clustering
Graph computation, GraphX in Spark supports common graph algorithms such as PageRank

The Spark on Kubernetes project is a solution for running Apache Spark applications on Kubernetes, providing a more cloud-native way to run them.

1.2 Iceberg

Iceberg is a table format, which we can define as a way of organizing data.

Iceberg is an open table format for huge analytic datasets, developed and open-sourced by Netflix. Iceberg adds tables using a high-performance format to Presto and Spark (Hudi also supports Presto and Spark integration), and the format works similarly to SQL tables.

The biggest difference from the underlying storage format (such as columnar storage formats like ORC and Parquet) is that it does not define how data is stored, but rather how data and metadata are organized, providing a unified table semantics upward.

The common usage is to create a table in iceberg format in Hive Metastore. Write to iceberg with flink or spark, and then read the table through other means, such as spark, flink, presto, etc. This article mainly adopts this technical route.

1.3 Hive Metastore

Hive Metastore is one of the core components of Apache Hive. It is mainly used as a metadata management service, providing a unified metadata storage and access layer for big data processing tools such as Hive, Spark, Presto, and Trino.

Simply put, Hive Metastore is responsible for managing and storing metadata such as table structures, database information, and storage locations, making it easy for distributed computing frameworks to quickly access and process large-scale data.

2. Deploying Hive Metastore

2.1 Deploying PG

Deploy with reference to the postgres15.yaml file in the https://github.com/shaowenchen/demo/tree/master/spark-3.5-iceberg directory.

Note that the PGSQL storage in the example uses local hostPath storage. For a production environment, it needs to be replaced with a more reliable storage service.

2.2 PG Database Initialization

Set environment variables

1
2
3
4
5
export POSTGRES_USER=postgresadmin
export POSTGRES_PASSWORD=postgrespassword
export POSTGRES_IP= 主机 IP
export POSTGRES_PORT= 集群 postgres svc 的 nodePort
export POSTGRES_DB=postgresdb

Create the database

1
psql -U $POSTGRES_USER -h $POSTGRES_IP -p $POSTGRES_PORT -d postgres -c "CREATE DATABASE $POSTGRES_DB;"

2.3 Deploying Hive Metastore

Deploy with reference to the hive-metastore.yaml file in the https://github.com/shaowenchen/demo/tree/master/spark-3.5-iceberg directory.

The S3-related bucket configuration needs to be replaced: BUCKET, ACCESS_KEY, SECRET_KEY, ACCESS_KEY, SECRET_KEY, etc.

Hive Metastore uses PGSQL as its database backend, and the PGSQL information is written in the deployment configuration file. If an external database is used, the relevant configuration in the deployment file needs to be updated.

3. Deploying Spark Operator

3.1 Introduction to Spark Operator

The figure above shows the architecture of Spark Operator and the relationships among its components. Its workflow is as follows:

spark-submit submits the Spark job to the Kubernetes cluster (it can be submitted via the sparkapplications object)
The Kubernetes cluster creates the Driver Pod
The Driver starts several Executor Pods
The Executor runs the concrete Task
After execution completes, the Driver cleans up the Executors

3.2 Installing Spark Operator

Add the repo

1
2
helm repo add spark-operator https://kubeflow.github.io/spark-operator
helm repo update

Install spark-operator

1
2
3
4
5
6
7
8
helm install spark-operator spark-operator/spark-operator \
    --version 2.0.0-rc.0 \
    --namespace spark-operator \
    --set 'spark.jobNamespaces={default,spark,spark-operator,spark}' \
    --create-namespace \
    --set webhook.enable=true \
    --set image.repository=kubeflow/spark-operator \
    --set image.tag=2.0.0-rc.0

spark-operator only processes Spark jobs in the namespaces specified by jobNamespaces.

Uninstall spark-operator

1
helm -n spark-operator uninstall spark-operator

View the workloads

1
2
3
4
5
kubectl -n spark-operator get pod

NAME                                         READY   STATUS    RESTARTS   AGE
spark-operator-controller-679bcc59c9-lsljx   1/1     Running   0          2d3h
spark-operator-webhook-676c675cdd-t8p95      1/1     Running   0          2d3h

View the CRDs

1
2
3
4
kubectl get crd |grep spark

scheduledsparkapplications.sparkoperator.k8s.io        2024-05-23T06:53:32Z
sparkapplications.sparkoperator.k8s.io                 2024-05-23T06:53:32Z

sparkapplications.sparkoperator.k8s.io defines a Spark job, and scheduledsparkapplications.sparkoperator.k8s.io defines a scheduled Spark job.

4. Processing Data with Standalone Spark and Iceberg

In standalone mode, Spark starts the complete set of dependencies locally.

4.1 Deploying a Standalone Spark Instance

Deploy with reference to the spark-iceberg.yaml file in the https://github.com/shaowenchen/demo/tree/master/spark-3.5-iceberg directory.

4.2 Entering the Spark Pod for Interactive Operations

1
kubectl -n spark exec -it spark-iceberg-749cf599dd-xdzlg bash

There are three available Spark interactive terminals:

spark-shell
spark-sql
spark-python

4.3 Creating an Iceberg Table

Run the spark-sql command to enter the spark-sql （default）> terminal

Create a namespace

1
CREATE NAMESPACE ns1;

Create a table

1
2
3
4
5
CREATE TABLE demo.ns1.table1 （
    id BIGINT,
    data STRING,
    category STRING
） USING iceberg;

The table created by the command above is saved to the spark-warehouse set in the default configuration file. If you want to specify the storage location, you can use the LOCATION keyword.

1
2
3
4
5
6
CREATE TABLE demo.ns1.table2 （
    id BIGINT,
    data STRING,
    category STRING
） USING iceberg
LOCATION 's3a://mybucket/datalake/spark-warehouse/mytable/';

4.4 Processing with spark-python

Exit the spark-shell terminal, save the code below into the Pod, and execute it with python spark-example.py.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
# spark-example.py
from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, length

sparkSessionConf = SparkConf（）.setAppName（f"DacnSpark Application [1]"）
spark = （
    SparkSession.builder.config（conf=sparkSessionConf）.enableHiveSupport（）.getOrCreate（）
）
if __name__ == "__main__":
    # 数据加载
    df = spark.read.json（
        "s3a://mybucket/datalake/spark-warehouse/source/ccnet-4000.json"
    ）
    # 数据处理
    df_cleaned = df.na.drop（）
    # 使用 iceberg 表管理，方便下一次处理
    df_cleaned.writeTo（"demo.ns1.table3"）.createOrReplace（）
    # 数据输出
    spark.read.table（"demo.ns1.table3"）.write.json（
        "s3a://mybucket/datalake/spark-warehouse/result/ccnet-4000-example.json", "overwrite"
    ）

This processing flow is divided into three parts:

Data loading: read external raw JSON data, or read data from an Iceberg table directly.
Data processing: use Spark’s computing power to process the data.
Data output: write the processed data to external storage S3 for business teams to use.

At this point, the Iceberg table’s data can be seen in object storage

The Iceberg table’s metadata

The data exported after processing

5. Processing Data in the Cluster with Spark

5.1 Submitting a Spark Job with YAML

Grant permissions to the Driver

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: spark-driver
rules:
  - apiGroups: [""]
    resources:
      [
        "pods",
        "configmaps",
        "secrets",
        "services",
        "persistentvolumeclaims",
        "events",
      ]
    verbs: ["*"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: spark-driver-binding
subjects:
  - kind: ServiceAccount
    name: default
    namespace: spark
roleRef:
  kind: ClusterRole
  name: spark-driver
  apiGroup: rbac.authorization.k8s.io

Submit the Spark job

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
  name: pyspark-pi
  namespace: spark
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: spark:3.5.1-python3
  imagePullPolicy: Always
  mainApplicationFile: local:///opt/spark/examples/src/main/python/pi.py
  sparkVersion: "3.5.1"
  restartPolicy:
    type: OnFailure
    onFailureRetries: 3
    onFailureRetryInterval: 10
    onSubmissionFailureRetries: 5
    onSubmissionFailureRetryInterval: 20
  driver:
    cores: 1
    coreLimit: "1200m"
    memory: "512m"
    labels:
      version: 3.5.1
    serviceAccount: default
  executor:
    cores: 1
    instances: 1
    memory: "512m"
    labels:
      version: 3.5.1

The official Spark image is used here, which has the examples built in.

Track the running status

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
kubectl -n spark get pod -w

NAME                             READY   STATUS              RESTARTS   AGE
pyspark-pi-driver                0/1     ContainerCreating   0          0s
spark-iceberg-749cf599dd-xdzlg   1/1     Running             0          20h
pyspark-pi-driver                1/1     Running             0          2s
pythonpi-5a8ac191caddfc56-exec-1   0/1     Pending             0          0s
pythonpi-5a8ac191caddfc56-exec-1   0/1     Pending             0          0s
pythonpi-5a8ac191caddfc56-exec-1   0/1     ContainerCreating   0          0s
pythonpi-5a8ac191caddfc56-exec-1   1/1     Running             0          2s
pythonpi-5a8ac191caddfc56-exec-1   1/1     Terminating         0          6s
pythonpi-5a8ac191caddfc56-exec-1   0/1     Terminating         0          7s
pythonpi-5a8ac191caddfc56-exec-1   0/1     Terminating         0          7s
pythonpi-5a8ac191caddfc56-exec-1   0/1     Terminating         0          7s
pyspark-pi-driver                  0/1     Completed           0          16s
pyspark-pi-driver                  0/1     Completed           0          18s

When it is Completed, the Spark job has finished; at the same time, the executor pods have been cleaned up, leaving only the driver pod.

View the result

1
2
3
kubectl -n spark logs pyspark-pi-driver -f |grep roughly

Pi is roughly 3.144760

The Driver obtains the execution result from the Executor and then outputs it locally.

Clean up

1
kubectl -n spark delete sparkapplication pyspark-pi

1
spark-submit --help

Basic usage

spark-submit [选项] <应用程序 JAR 文件 | Python 文件 | R 文件> [应用程序参数]
spark-submit --kill [提交 ID] --master [spark://...]
spark-submit --status [提交 ID] --master [spark://...]
spark-submit run-example [选项] 示例类名 [示例参数]

Options

–master MASTER_URL: the master URL of the Spark cluster (such as spark://host:port, mesos://host:port, yarn, k8s://https://host:port, or local mode local[*]), defaulting to local[*] (that is, running with all local CPU cores).

–deploy-mode DEPLOY_MODE: specifies the deployment mode of the driver, “client” means running the driver locally, “cluster” means running the driver in the cluster (defaults to client mode).

–class CLASS_NAME: the main class of your application (for Java / Scala applications).

–name NAME: the name of the application.

–jars JARS: a comma-separated list of JAR files to include in the classpaths of the driver and executors.

–packages: a comma-separated list of Maven coordinates specifying the JAR packages to include in the classpaths of the driver and executors. It searches the local Maven repository first, then Maven Central and other remote repositories specified via --repositories.

–exclude-packages: a comma-separated list of groupId:artifactId used to exclude specified dependency packages, avoiding conflicts with dependencies provided by --packages.

–repositories: a comma-separated list of remote repositories used to search for the Maven coordinates specified by --packages.

–py-files PY_FILES: a comma-separated list of .zip, .egg, or .py files that will be placed on the PYTHONPATH of the Python application.

–files FILES: a comma-separated list of files that will be placed in each executor’s working directory; these files can be accessed via SparkFiles.get（fileName）.

–archives ARCHIVES: a comma-separated list of archive files that will be extracted into each executor’s working directory.

–conf, -c PROP=VALUE: set arbitrary Spark configuration properties.

–properties-file FILE: specify a file path from which to load additional configuration properties. If not specified, Spark looks for the conf/spark-defaults.conf file by default.

–driver-memory MEM: specifies the driver’s memory size (such as 1000M, 2G), defaulting to 1024M.

–driver-java-options: pass additional Java options to the driver.

–driver-library-path: pass an additional library path to the driver.

–driver-class-path: pass an additional classpath to the driver. Note that JAR files added with --jars are automatically included in the classpath.

–executor-memory MEM: specifies the memory size of each executor (such as 1000M, 2G), defaulting to 1G.

–proxy-user NAME: specifies the user to impersonate when submitting the application. This option is incompatible with --principal and --keytab.

–help, -h: show help information and exit.

–verbose, -v: print additional debug information.

–version: print the current Spark version information.

Spark Connect (Spark Connect mode only)

–remote CONNECT_URL: specifies the URL for connecting to the Spark Connect service, for example sc://host:port. This option cannot be set together with --master and --deploy-mode. This option is experimental and may change in minor releases.

Cluster Deploy Mode Only

–driver-cores NUM: specifies the number of cores used by the driver (only used in cluster mode, defaults to 1).

Spark Standalone or Mesos (cluster mode only)

–supervise: if this option is specified, the driver is automatically restarted on failure.

Spark Standalone, Mesos, or K8S (cluster mode only)

–kill SUBMISSION_ID: when specified, kills the driver with the given ID.

–status SUBMISSION_ID: when specified, gets the status of the driver with the given ID.

Spark Standalone and Mesos (Standalone and Mesos modes only)

–total-executor-cores NUM: specifies the total number of cores used by all executors.

Spark Standalone, YARN, and Kubernetes (Standalone, YARN, and K8S modes only)

–executor-cores NUM: the number of cores used by each executor (defaults to 1 in YARN and Kubernetes modes, and defaults to all cores available on the worker node in Standalone mode).

Spark on YARN and Kubernetes (YARN and K8S modes only)

–num-executors NUM: the number of executors to launch (defaults to 2). If dynamic resource allocation is enabled, the initial number of executors is at least the specified value.

–principal PRINCIPAL: specifies the principal used to log in to the KDC.

–keytab KEYTAB: specifies the full path of the keytab file corresponding to the principal.

Spark on YARN (YARN mode only)

–queue QUEUE_NAME: specifies the YARN queue name, defaulting to “default”.

5.3 Submitting a Spark Job with spark-submit

Enter the Pod terminal

1
kubectl -n spark exec -it spark-iceberg-749cf599dd-xdzlg bash

Submit the Spark job

In the steps above, sufficient permissions have already been granted to the default ServiceAccount, so here we just need to specify the master address.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
/opt/spark/bin/spark-submit \
  --master k8s://https://x.x.x.x:6443 \
  --deploy-mode cluster \
  --name spark-iceberg-example \
  --class org.apache.spark.deploy.python.PythonRunner \
  --conf spark.executor.instances=2 \
  --conf spark.executor.memory=16G \
  --conf spark.executor.cores=8 \
  --conf spark.driver.memory=16g \
  --conf spark.pyspark.python=/usr/bin/python3 \
  --conf spark.kubernetes.container.image.pullPolicy=Always \
  --conf spark.kubernetes.container.image=shaowenchen/spark:3.5.1-python3-s3 \
  --conf spark.eventLog.enabled=false \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=default \
  --conf spark.hadoop.fs.s3a.bucket.name=mybucket \
  --conf spark.hadoop.fs.s3a.access.key=xxx \
  --conf spark.hadoop.fs.s3a.secret.key=xxx \
  --conf spark.kubernetes.file.upload.path=s3a://mybucket/datalake/spark-warehouse/upload \
  /opt/spark/examples/src/main/python/pi.py

The --conf entries contain a large number of configuration parameters; see https://spark.apache.org/docs/latest/running-on-kubernetes.html for details.

During execution, information about the interaction with Kubernetes is printed, mainly the status of the Pod

View the related Pods

1
2
3
spark-iceberg-example-e611cd91cb54a5fc-driver   0/1     Completed   0          50s
pythonpi-e80edc91cb54c2eb-exec-1                0/1     Terminating         0          5s
pythonpi-e80edc91cb54c2eb-exec-2                0/1     Terminating         0          5s

As above, the driver is retained after it finishes executing, and the executors are deleted.

View the related logs

1
2
3
kubectl -n spark logs spark-iceberg-example-e611cd91cb54a5fc-driver -f |grep roughly

Pi is roughly 3.144840

6. Processing Data in the Cluster with Spark and Iceberg

Since the scripts that Spark processes may be adjusted from time to time in a production environment, in order to make scripts easier to update and manage, the scripts for Spark data processing need to be mounted into the Driver and Executor via a PVC.

See https://github.com/shaowenchen/demo/blob/master/spark-3.5-iceberg/sparkapp.yaml

7. Exposing a Spark Processing API with Argo Webhook

Create a Sensor

https://github.com/shaowenchen/demo/blob/master/spark-3.5-iceberg/argo-sensor.yaml

Create an EventSource

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: spark-webhook
  namespace: argo
spec:
  service:
    ports:
      - port: 13000
        targetPort: 13000
  webhook:
    sparkapp-start:
      port: "13000"
      endpoint: /sparkapp/start
      method: POST
      url: ""

Check the Sensor Pod status

1
2
3
kubectl -n argo get pod  |grep spark

sparkapp-start-sensor-sensor-gkxzn-58d7648fd7-fs8np               1/1     Running     0             22s

View the Webhook’s service port

1
2
3
4
5
kubectl -n argo get svc

NAME                                  TYPE        CLUSTER-IP     EXTERNAL-IP   PORT（S）                      AGE
eventbus-default-stan-svc             ClusterIP   None           <none>        4222/TCP,6222/TCP,8222/TCP   291d
spark-webhook-eventsource-svc         ClusterIP   10.103.215.1   <none>        13000:30001/TCP                    291d

Call the API to trigger the task

1
curl -d '{"script":"spark-example.py", "pvc":"mypvc", "path":"spark", "executor":2, "ak":"xxx", "sk":"xxx", "endpoint":"ks3-cn-beijing-internal.ksyuncs.com", "bucket": "mybucket"}' -H "Content-Type: application/json" -X POST http://localhost:31300/sparkapp/start -v

The script to start can be specified via script, and path can be used to isolate directories within the PVC.

8. Summary

This article records the process of deploying a data processing software stack based on Spark, Iceberg, and Hive Metastore. The main contents are as follows:

Introduced the basic concepts of Spark, Iceberg, and Hive Metastore
Deployed Hive Metastore and Spark Operator
Tested running Spark jobs in standalone mode
Tested running Spark jobs with spark-submit and YAML
Exposed a Spark data processing API externally via Argo Webhook