Deploy Layanan Remote Shuffle Celeborn untuk Spark di ACK - Container Service for Kubernetes

Apache Celeborn adalah layanan yang dirancang untuk mengelola data perantara, seperti shuffle data dan spill data, pada mesin komputasi data besar. Sebagai remote shuffle service (RSS), layanan ini secara efisien menangani proses shuffle untuk dataset berskala besar, sehingga meningkatkan kinerja, stabilitas, dan fleksibilitas sistem. Topik ini menjelaskan cara menerapkan komponen Celeborn di kluster Container Service for Kubernetes (ACK) dan menggunakannya sebagai RSS untuk pekerjaan Spark.

Manfaat

Untuk kerangka kerja pemrosesan data besar seperti MapReduce, Spark, dan Flink, penggunaan Celeborn sebagai RSS memberikan manfaat berikut:

Penulisan shuffle berbasis push: Node mapper tidak perlu menyimpan data di disk lokal, sehingga ideal untuk arsitektur cloud-native dengan storage and compute separation.
Pembacaan shuffle berbasis merge: Data digabungkan di node worker alih-alih di node reducer. Hal ini menghindari I/O file kecil acak dan beban jaringan akibat transfer data kecil, sehingga meningkatkan efisiensi pemrosesan data.
high availability: Node master Celeborn menggunakan protokol konsensus Raft untuk mencapai high availability dan memastikan stabilitas sistem.
fault tolerance: Celeborn mendukung dual replicas, yang secara signifikan mengurangi kemungkinan kegagalan fetch.

Prasyarat

Komponen ack-spark-operator telah diinstal. Untuk informasi selengkapnya, lihat Langkah 1: Terapkan komponen ack-spark-operator.
Klien kubectl telah terhubung ke kluster ACK. Untuk informasi selengkapnya, lihat Hubungkan ke kluster ACK menggunakan kubectl.
Anda telah membuat dan mengonfigurasi node pools sesuai dengan lingkungan kluster yang dijelaskan pada bagian berikut. Untuk informasi selengkapnya, lihat Buat dan kelola node pool.

Lingkungan kluster

Contoh ini menggunakan konfigurasi kluster ACK berikut:

Proses master diterapkan ke node pool celeborn-master dengan konfigurasi berikut:
- Nama node pool: celeborn-master
- Jumlah node: 3
- ECS instance type: g8i.2xlarge
- label: celeborn.apache.org/role=master
- taint: celeborn.apache.org/role=master:NoSchedule
- Penyimpanan data per node: /mnt/celeborn_ratis (1024 GB)
Proses worker diterapkan ke node pool celeborn-worker dengan konfigurasi berikut:
- Nama node pool: celeborn-worker
- Jumlah node: 5
- ECS instance type: g8i.4xlarge
- label: celeborn.apache.org/role=worker
- taint: celeborn.apache.org/role=worker:NoSchedule
- Penyimpanan data per node:
  - /mnt/disk1 (1024 GB)
  - /mnt/disk2 (1024 GB)
  - /mnt/disk3 (1024 GB)
  - /mnt/disk4 (1024 GB)

Ikhtisar prosedur

Panduan ini memandu Anda melalui langkah-langkah untuk menerapkan Celeborn di kluster ACK.

Buat container image Celeborn

Unduh release Celeborn, buat container image, lalu dorong ke image repository Anda untuk menerapkan komponen ack-celeborn.
Terapkan komponen ack-celeborn

Gunakan Helm chart ack-celeborn dari Marketplace ACK untuk menerapkan kluster Celeborn menggunakan container image yang telah Anda buat.
Buat container image Spark

Buat container image Spark yang mencakup dependensi untuk Celeborn dan akses ke OSS, lalu dorong image tersebut ke image repository Anda.
Siapkan dan unggah data uji ke OSS

Hasilkan set data uji untuk pekerjaan PageRank dan unggah ke OSS.
Jalankan pekerjaan Spark contoh

Jalankan pekerjaan PageRank contoh dan konfigurasikan agar menggunakan Celeborn sebagai RSS.
(Opsional) Bersihkan resource

Setelah menyelesaikan tutorial ini, bersihkan pekerjaan Spark dan resource lain yang tidak diperlukan untuk menghindari biaya tambahan.

Langkah 1: Buat container image Celeborn

Unduh release yang diperlukan (misalnya, versi 0.5.2) dari situs web resmi Celeborn. Saat mengonfigurasi, ganti <IMAGE-REGISTRY> dan <IMAGE-REPOSITORY> dengan registri dan nama image Anda. Anda juga dapat mengubah variabel PLATFORMS untuk mengonfigurasi arsitektur image yang diperlukan. Untuk informasi selengkapnya, lihat Deploy Celeborn on Kubernetes. Perintah docker buildx memerlukan Docker 19.03 atau versi lebih baru. Untuk informasi tentang cara meningkatkan Docker, lihat Instal dan gunakan Docker dan Docker Compose.

CELEBORN_VERSION=0.5.2               # Versi Celeborn.

IMAGE_REGISTRY=<IMAGE-REGISTRY>      # Registri image, misalnya docker.io.

IMAGE_REPOSITORY=<IMAGE-REPOSITORY>  # Nama image, misalnya apache/celeborn.

IMAGE_TAG=${CELEBORN_VERSION}        # Tag image. Contoh ini menggunakan versi Celeborn sebagai tag.

PLATFORMS=linux/amd64                # Arsitektur platform image. Untuk mendukung beberapa platform, pisahkan dengan koma, misalnya linux/amd64,linux/arm64.

# Unduh paket rilis.
wget https://downloads.apache.org/celeborn/celeborn-${CELEBORN_VERSION}/apache-celeborn-${CELEBORN_VERSION}-bin.tgz

# Ekstrak paket.
tar -zxvf apache-celeborn-${CELEBORN_VERSION}-bin.tgz

# Masuk ke direktori kerja.
cd apache-celeborn-${CELEBORN_VERSION}-bin

# Gunakan Docker Buildx untuk membuat image dan mendorongnya ke repositori image.
docker buildx build \
    --output=type=registry \
    --push \
    --platform=${PLATFORMS} \
    --tag=${IMAGE_REGISTRY}/${IMAGE_REPOSITORY}:${IMAGE_TAG} \
    -f docker/Dockerfile \
    .

Langkah 2: Terapkan komponen ack-celeborn

Masuk ke Konsol ACK. Di panel navigasi kiri, klik Marketplace > Marketplace.
Di halaman Marketplace, klik tab App Catalog, temukan dan pilih ack-celeborn, lalu di halaman ack-celeborn, klik Deploy.
Di panel Create, pilih kluster dan namespace, lalu klik Next.

Di halaman Parameters, konfigurasikan parameter, lalu klik OK.

image:                         # Ganti ini dengan alamat image Celeborn yang Anda buat di Langkah 1.
  registry: docker.io          # Registri image.
  repository: apache/celeborn  # Nama image.
  tag: 0.5.2                   # Tag image.

celeborn:
  celeborn.client.push.stageEnd.timeout: 120s
  celeborn.master.ha.enabled: true
  celeborn.master.ha.ratis.raft.server.storage.dir: /mnt/celeborn_ratis
  celeborn.master.heartbeat.application.timeout: 300s
  celeborn.master.heartbeat.worker.timeout: 120s
  celeborn.master.http.port: 9098
  celeborn.metrics.enabled: true
  celeborn.metrics.prometheus.path: /metrics/prometheus
  celeborn.rpc.dispatcher.numThreads: 4
  celeborn.rpc.io.clientThreads: 64
  celeborn.rpc.io.numConnectionsPerPeer: 2
  celeborn.rpc.io.serverThreads: 64
  celeborn.shuffle.chunk.size: 8m
  celeborn.worker.fetch.io.threads: 32
  celeborn.worker.flusher.buffer.size: 256K
  celeborn.worker.http.port: 9096
  celeborn.worker.monitor.disk.enabled: false
  celeborn.worker.push.io.threads: 32
  celeborn.worker.storage.dirs: /mnt/disk1:disktype=SSD:capacity=1024Gi,/mnt/disk2:disktype=SSD:capacity=1024Gi,/mnt/disk3:disktype=SSD:capacity=1024Gi,/mnt/disk4:disktype=SSD:capacity=1024Gi

master:
  replicas: 3
  env:
  - name: CELEBORN_MASTER_MEMORY
    value: 28g
  - name: CELEBORN_MASTER_JAVA_OPTS
    value: -XX:-PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -Xloggc:gc-master.out -Dio.netty.leakDetectionLevel=advanced
  - name: CELEBORN_NO_DAEMONIZE
    value: "1"
  - name: TZ
    value: Asia/Shanghai
  volumeMounts:
  - name: celeborn-ratis
    mountPath: /mnt/celeborn_ratis
  resources:
    requests:
      cpu: 7                
      memory: 28Gi        
    limits:
      cpu: 7
      memory: 28Gi
  volumes:
  - name: celeborn-ratis
    hostPath:
      path: /mnt/celeborn_ratis
      type: DirectoryOrCreate
  nodeSelector:
    celeborn.apache.org/role: master
  tolerations:
  - key: celeborn.apache.org/role
    operator: Equal
    value: master
    effect: NoSchedule

worker:
  replicas: 5
  env:
  - name: CELEBORN_WORKER_MEMORY
    value: 28g
  - name: CELEBORN_WORKER_OFFHEAP_MEMORY
    value: 28g
  - name: CELEBORN_WORKER_JAVA_OPTS
    value: -XX:-PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -Xloggc:gc-worker.out -Dio.netty.leakDetectionLevel=advanced
  - name: CELEBORN_NO_DAEMONIZE
    value: "1"
  - name: TZ
    value: Asia/Shanghai
  volumeMounts:
  - name: disk1
    mountPath: /mnt/disk1
  - name: disk2
    mountPath: /mnt/disk2
  - name: disk3
    mountPath: /mnt/disk3
  - name: disk4
    mountPath: /mnt/disk4
  resources:
    requests:
      cpu: 14
      memory: 56Gi
    limits:
      cpu: 14
      memory: 56Gi
  volumes:
  - name: disk1
    hostPath:
      path: /mnt/disk1
      type: DirectoryOrCreate
  - name: disk2
    hostPath:
      path: /mnt/disk2
      type: DirectoryOrCreate
  - name: disk3
    hostPath:
      path: /mnt/disk3
      type: DirectoryOrCreate
  - name: disk4
    hostPath:
      path: /mnt/disk4
      type: DirectoryOrCreate
  nodeSelector:
    celeborn.apache.org/role: worker
  tolerations:
  - key: celeborn.apache.org/role
    operator: Equal
    value: worker
    effect: NoSchedule

Tabel berikut menjelaskan parameter utama. Untuk daftar lengkap parameter, lihat bagian ConfigMaps di halaman ack-celeborn.

Pengaturan parameter

Parameter	Deskripsi	Contoh
`image.registry`	Alamat `image registry`.	`"docker.io"`
`image.repository`	Nama image.	`"apache/celeborn"`
`image.tag`	Tag image.	`"0.5.2"`
`image.pullPolicy`	Kebijakan pull image.	`"IfNotPresent"`
`celeborn`	Properti konfigurasi Celeborn.	{ "celeborn.client.push.stageEnd.timeout": "120s", "celeborn.master.ha.enabled": true, "celeborn.master.ha.ratis.raft.server.storage.dir": "/mnt/celeborn_ratis", "celeborn.master.heartbeat.application.timeout": "300s", "celeborn.master.heartbeat.worker.timeout": "120s", "celeborn.master.http.port": 9098, "celeborn.metrics.enabled": true, "celeborn.metrics.prometheus.path": "/metrics/prometheus", "celeborn.rpc.dispatcher.numThreads": 4, "celeborn.rpc.io.clientThreads": 64, "celeborn.rpc.io.numConnectionsPerPeer": 2, "celeborn.rpc.io.serverThreads": 64, "celeborn.shuffle.chunk.size": "8m", "celeborn.worker.fetch.io.threads": 32, "celeborn.worker.flusher.buffer.size": "256K", "celeborn.worker.http.port": 9096, "celeborn.worker.monitor.disk.enabled": false, "celeborn.worker.push.io.threads": 32, "celeborn.worker.storage.dirs": "/mnt/disk1:disktype=SSD:capacity=1024Gi,/mnt/disk2:disktype=SSD:capacity=1024Gi,/mnt/disk3:disktype=SSD:capacity=1024Gi,/mnt/disk4:disktype=SSD:capacity=1024Gi" }
`master.replicas`	Jumlah replika `pod` master.	`3`
`master.volumeMounts`	`volumeMounts` untuk kontainer master.	`[ { "mountPath": "/mnt/celeborn_ratis", "name": "celeborn-ratis" } ]`
`master.volumes`	Volume data untuk `pod` master. Hanya tipe volume `hostPath` dan `emptyDir` yang didukung.	`[ { "hostPath": { "path": "/mnt/celeborn_ratis", "type": "DirectoryOrCreate" }, "name": "celeborn-ratis" } ]`
`master.nodeSelector`	`node selector` untuk `pod` master.	`{}`
`master.affinity`	Aturan `affinity` untuk `pod` master.	`{ "podAntiAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": [ { "labelSelector": { "matchExpressions": [ { "key": "app.kubernetes.io/name", "operator": "In", "values": [ "celeborn" ] }, { "key": "app.kubernetes.io/role", "operator": "In", "values": [ "master" ] } ] }, "topologyKey": "kubernetes.io/hostname" } ] } }`
`master.tolerations`	`tolerations` untuk `pod` master.	`[]`
`worker.replicas`	Jumlah replika `pod` `worker`.	`5`
`worker.volumeMounts`	`volumeMounts` untuk kontainer `worker`.	`[ { "mountPath": "/mnt/disk1", "name": "disk1" }, { "mountPath": "/mnt/disk2", "name": "disk2" }, { "mountPath": "/mnt/disk3", "name": "disk3" }, { "mountPath": "/mnt/disk4", "name": "disk4" } ]`
`worker.volumes`	Volume data untuk `pod` `worker`. Hanya tipe volume `hostPath` dan `emptyDir` yang didukung.	`[ { "capacity": "100Gi", "diskType": "SSD", "hostPath": "/mnt/disk1", "mountPath": "/mnt/disk1", "type": "hostPath" }, { "capacity": "100Gi", "diskType": "SSD", "hostPath": "/mnt/disk2", "mountPath": "/mnt/disk2", "type": "hostPath" }, { "capacity": "100Gi", "diskType": "SSD", "hostPath": "/mnt/disk3", "mountPath": "/mnt/disk3", "type": "hostPath" }, { "capacity": "100Gi", "diskType": "SSD", "hostPath": "/mnt/disk4", "mountPath": "/mnt/disk4", "type": "hostPath" } ]`
`worker.nodeSelector`	`node selector` untuk `pod` `worker`.	`{}`
`worker.affinity`	Aturan `affinity` untuk `pod` `worker`.	`{ "podAntiAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": [ { "labelSelector": { "matchExpressions": [ { "key": "app.kubernetes.io/name", "operator": "In", "values": [ "celeborn" ] }, { "key": "app.kubernetes.io/role", "operator": "In", "values": [ "worker" ] } ] }, "topologyKey": "kubernetes.io/hostname" } ] } }`
`worker.tolerations`	`tolerations` untuk `pod` `worker`.	`[]`

Jalankan perintah berikut dan tunggu hingga penerapan Celeborn selesai. Jika Anda mengalami masalah pod selama penerapan komponen, lihat Troubleshooting Pod.
```
kubectl get -n celeborn statefulset 
```
Output yang diharapkan:
```
NAME              READY   AGE
celeborn-master   3/3     68s
celeborn-worker   5/5     68s
```

Langkah 3: Buat container image Spark

Contoh ini menggunakan Spark 3.5.3. Buat Dockerfile dengan konten berikut untuk membuat container image dan dorong ke image repository Anda.

ARG SPARK_IMAGE=<SPARK_IMAGE>  # Ganti <SPARK_IMAGE> dengan image dasar Spark Anda.

FROM ${SPARK_IMAGE}

# Tambahkan dependensi untuk dukungan Hadoop Aliyun OSS
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aliyun/3.3.4/hadoop-aliyun-3.3.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/com/aliyun/oss/aliyun-sdk-oss/3.17.4/aliyun-sdk-oss-3.17.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/jdom/jdom2/2.0.6.1/jdom2-2.0.6.1.jar ${SPARK_HOME}/jars

# Tambahkan dependensi untuk Celeborn
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/celeborn/celeborn-client-spark-3-shaded_2.12/0.5.1/celeborn-client-spark-3-shaded_2.12-0.5.1.jar ${SPARK_HOME}/jars

Langkah 4: Unggah data uji ke OSS

Untuk informasi tentang cara menyiapkan data uji dan mengunggahnya ke OSS, lihat Langkah 1: Siapkan data uji dan unggah ke OSS.

Langkah 5: Buat Secret OSS

Untuk informasi tentang cara membuat Secret untuk menyimpan kredensial akses OSS, lihat Langkah 3: Buat Secret untuk menyimpan kredensial akses OSS.

Langkah 6: Kirim pekerjaan Spark contoh

Buat file manifest SparkApplication bernama spark-pagerank.yaml dengan konten berikut. Ganti <SPARK_IMAGE> dengan nama image yang Anda buat di Langkah 3: Buat container image Spark, dan ganti <OSS_BUCKET> serta <OSS_ENDPOINT> dengan Bucket OSS dan Titik akhir Anda. Untuk informasi selengkapnya tentang cara mengonfigurasi Celeborn dalam pekerjaan Spark, lihat dokumentasi Celeborn.

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: <SPARK_IMAGE>                                     # Image Spark. Ganti <SPARK_IMAGE> dengan nama image Spark Anda.
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.3.jar
  mainClass: org.apache.spark.examples.SparkPageRank
  arguments:
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt           # Set data uji input. Ganti <OSS_BUCKET> dengan nama bucket OSS Anda.
  - "10"                                                   # Jumlah iterasi.
  sparkVersion: 3.5.3
  hadoopConf:
    fs.AbstractFileSystem.oss.impl: org.apache.hadoop.fs.aliyun.oss.OSS
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    fs.oss.endpoint: <OSS_ENDPOINT>                        # Titik akhir OSS. Misalnya, titik akhir internal untuk OSS di wilayah China (Beijing) adalah oss-cn-beijing-internal.aliyuncs.com. 
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  sparkConf:
    spark.shuffle.manager: org.apache.spark.shuffle.celeborn.SparkShuffleManager
    spark.serializer: org.apache.spark.serializer.KryoSerializer
    spark.celeborn.master.endpoints: celeborn-master-0.celeborn-master-svc.celeborn.svc.cluster.local,celeborn-master-1.celeborn-master-svc.celeborn.svc.cluster.local,celeborn-master-2.celeborn-master-svc.celeborn.svc.cluster.local
    spark.celeborn.client.spark.shuffle.writer: hash
    spark.celeborn.client.push.replicate.enabled: "false"
    spark.sql.adaptive.localShuffleReader.enabled: "false"
    spark.sql.adaptive.enabled: "true"
    spark.sql.adaptive.skewJoin.enabled: "true"
    spark.shuffle.sort.io.plugin.class: org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO
    spark.dynamicAllocation.shuffleTracking.enabled: "false"
    spark.executor.userClassPathFirst: "false"
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    serviceAccount: spark-operator-spark
    envFrom:
    - secretRef:
        name: spark-oss-secret
  executor:
    instances: 2
    cores: 1
    coreLimit: "2"
    memory: 8g
    envFrom:
    - secretRef:
        name: spark-oss-secret
  restartPolicy:
    type: Never

(Opsional) Langkah 7: Bersihkan resource

Setelah menyelesaikan tutorial ini, jalankan perintah berikut untuk menghapus resource dan menghindari biaya tambahan.

Jalankan perintah berikut untuk menghapus pekerjaan Spark:

kubectl delete sparkapplication spark-pagerank

Jalankan perintah berikut untuk menghapus resource Secret:

kubectl delete secret spark-oss-secret

Referensi

Untuk informasi tentang cara menggunakan Spark Operator untuk mengirim pekerjaan Spark, lihat Gunakan Spark Operator untuk menjalankan pekerjaan Spark.
Untuk informasi tentang cara menggunakan Spark History Server untuk melihat informasi pekerjaan Spark, lihat Gunakan Spark History Server untuk melihat informasi tentang pekerjaan Spark.
Untuk informasi tentang cara menggunakan Celeborn, lihat dokumentasi Apache Celeborn.