全部產品
Search

搜尋本產品

    Container Service for Kubernetes：Spark作業使用Celeborn作為RSS

    文件中心

    Container Service for Kubernetes：Spark作業使用Celeborn作為RSS

    更新時間：Mar 13, 2025

    Apache Celeborn是一個專門用於處理巨量資料計算引擎中間資料（如Shuffle資料和溢寫資料）的服務，能夠提升巨量資料引擎的效能、穩定性和靈活性。Remote Shuffle Service（RSS）用於高效處理大規模資料集的Shuffle過程。本文介紹如何在ACK叢集中部署Celeborn組件，並在Spark作業中使用Celeborn作為Remote Shuffle Service（RSS）。

    在Spark作業中使用Celeborn的優勢

    對於MapReduce、Spark和Flink等巨量資料處理架構，使用Celeborn作為RSS具有如下優勢：

    • 推送式Shuffle寫入（Push-based shuffle write）：Mapper節點不需要將資料存放區在本地磁碟，適合雲端存算分離架構。

    • 合并式Shuffle讀取（Merge-based shuffle read）：資料在Worker節點進行合并，而非在Reducer節點，避免小檔案的隨機讀寫及小資料量傳輸帶來的網路開銷，提升資料處理效率。

    • 高可用性：Celeborn的Master節點基於Raft協議實現高可用性，確保系統的穩定運行。

    • 高容錯性：支援雙副本機制，顯著降低Fetch失敗的機率。

    前提條件

    叢集環境

    本樣本中使用的ACK叢集環境資訊如下所示。

    • Master進程部署到節點池celeborn-master中，配置如下：

      • 節點池名稱：celeborn-master

      • 節點數：3

      • ECS執行個體規格類型：g8i.2xlarge

      • 標籤：celeborn.apache.org/role=master

      • 汙點：celeborn.apache.org/role=master:NoSchedule

      • 單節點資料存放區：/mnt/celeborn_ratis(1024GB）

    • Worker進程部署到節點池celeborn-worker中，配置如下：

      • 節點池名稱：celeborn-worker

      • 節點數：5

      • ECS執行個體規格類型：g8i.4xlarge

      • 標籤：celeborn.apache.org/role=worker

      • 汙點：celeborn.apache.org/role=worker:NoSchedule

      • 單節點資料存放區：

        • /mnt/disk1(1024GB）

        • /mnt/disk2(1024GB）

        • /mnt/disk3(1024GB）

        • /mnt/disk4(1024GB）

    流程概述

    本文將引導您完成以下步驟，協助您瞭解如何在ACK叢集中部署Celeborn。

    1. 構建Celeborn容器鏡像

      根據所需的Celeborn版本下載相應的發行版，然後構建容器鏡像並將其推送至您的鏡像倉庫，以供部署ack-celeborn組件時使用。

    2. 部署ack-celeborn組件

      通過ACK應用市場提供的ack-celeborn Helm Chart，使用已構建的Celeborn容器鏡像，一鍵部署Celeborn叢集。

    3. 構建Spark容器鏡像

      構建包含了Celeborn和訪問OSS相關Jar包依賴的Spark容器鏡像，並推送到您的鏡像倉庫中。

    4. 準備測試資料並上傳至OSS

      產生PageRank作業的測試資料集並將其上傳至OSS。

    5. 運行樣本Spark作業

      運行樣本PageRank作業並配置使用Celeborn作為RSS。

    6. （可選）環境清理

      體驗完本教程後，清理無需使用的Spark作業和資源，避免產生額外的費用。

    步驟一：構建Celeborn容器鏡像

    根據您所使用的Celeborn版本，從Celeborn 官網下載相應的發行版（如0.5.2版本）。在配置過程中，將<IMAGE-REGISTRY>和<IMAGE-REPOSITORY>替換為您自己的鏡像倉庫和鏡像名稱。同時，您可以通過修改PLATFORMS變數來配置所需的鏡像架構。更多資訊，請參見Deploy Celeborn on Kubernetes。docker buildx命令需要Docker版本19.03或更高版本支援，升級詳情請參見安裝Docker。

    CELEBORN_VERSION=0.5.2               # Celeborn版本號碼。

IMAGE_REGISTRY=<IMAGE-REGISTRY>      # 鏡像倉庫，例如docker.io。

IMAGE_REPOSITORY=<IMAGE-REPOSITORY>  # 鏡像名稱，例如apache/celeborn。

IMAGE_TAG=${CELEBORN_VERSION}        # 鏡像標籤，這裡使用Celeborn版本號碼作為標籤。

# 下載。
wget https://downloads.apache.org/celeborn/celeborn-${CELEBORN_VERSION}/apache-celeborn-${CELEBORN_VERSION}-bin.tgz

# 解壓。
tar -zxvf apache-celeborn-${CELEBORN_VERSION}-bin.tgz

# 切換工作目錄。
cd apache-celeborn-${CELEBORN_VERSION}-bin

# 使用Docker Buildkit構建鏡像並推送到鏡像倉庫中。
docker buildx build \
    --output=type=registry \
    --push \
    --platform=${PLATFORMS} \
    --tag=${IMAGE_REGISTRY}/${IMAGE_REPOSITORY}:${IMAGE_TAG} \
    -f docker/Dockerfile \
    .

    步驟二：部署ack-celeborn組件

    1. 登入Container Service管理主控台，在左側導覽列選擇市場 > 應用市場。

    2. 在應用市場頁面，單擊應用目錄頁簽，然後搜尋並選中ack-celeborn，然後在ack-celeborn頁面，單擊一鍵部署。

    3. 在建立面板中，選擇叢集和命名空間，然後單擊下一步。

    4. 在參數配置頁面，設定相應參數，然後單擊確定。

      image:                         # 需替換成步驟一中構建得到的Celeborn鏡像地址。
  registry: docker.io          # 鏡像倉庫。
  repository: apache/celeborn  # 鏡像名稱。
  tag: 0.5.2                   # 鏡像標籤。

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

      下表列出了部分配置參數的說明。完整的參數配置詳情，您可以在ack-celeborn頁面中的配置項查看。

      參數配置

      參數

      描述

      樣本值

      image.registry

      鏡像倉庫地址。

      "docker.io"

      image.repository

      鏡像名稱。

      "apache/celeborn"

      image.tag

      鏡像標籤。

      "0.5.1"

      image.pullPolicy

      鏡像拉取策略。

      "IfNotPresent"

      celeborn

      Celeoborn配置項。

      {
  "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

      Master Pod副本數量。

      3

      master.volumeMounts

      Master 容器資料卷掛載。

      [
  {
    "mountPath": "/mnt/celeborn_ratis",
    "name": "celeborn-ratis"
  }
]

      master.volumes

      Master Pod資料卷。

      目前資料卷類型僅支援hostPath和emptyDir類型。

      [
  {
    "hostPath": {
      "path": "/mnt/celeborn_ratis",
      "type": "DirectoryOrCreate"
    },
    "name": "celeborn-ratis"
  }
]

      master.nodeSelector

      Master Pod節點選取器。

      {}

      master.affinity

      Master Pod親和性。

      {
  "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

      Master Pod汙點容忍。

      []

      worker.replicas

      Worker Pod副本數量。

      5

      worker.volumeMounts

      Worker 容器資料卷掛載。

      [
  {
    "mountPath": "/mnt/disk1",
    "name": "disk1"
  },
  {
    "mountPath": "/mnt/disk2",
    "name": "disk2"
  },
  {
    "mountPath": "/mnt/disk3",
    "name": "disk3"
  },
  {
    "mountPath": "/mnt/disk4",
    "name": "disk4"
  }
]

      worker.volumes

      Worker Pod資料卷。

      目前資料卷類型僅支援hostPath和emptyDir類型；

      [
  {
    "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

      Worker Pod節點選取器。

      {}

      worker.affinity

      Worker Pod親和性。

      {
  "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

      Worker Pod汙點容忍。

      []

    1. 執行以下命令並耐心等待Celeborn部署完成。在組件部署期間，如遇到Pod異常問題請參見Pod異常問題排查。

      kubectl get -n celeborn statefulset

      預期輸出：

      NAME              READY   AGE
celeborn-master   3/3     68s
celeborn-worker   5/5     68s

    步驟三：構建Spark容器鏡像

    以Spark 3.5.3版本為例，建立如下Dockerfile檔案，構建並上傳至您的鏡像倉庫。

    ARG SPARK_IMAGE=<SPARK_IMAGE>  # 將<SPARK_IMAGE>替換成您的Spark基礎鏡像。

FROM ${SPARK_IMAGE}

# Add dependency for Hadoop Aliyun OSS support
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

# Add dependency for 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

    步驟四：準備測試資料並上傳至 OSS

    關於如何準備測試資料並上傳至OSS，參見步驟一：準備測試資料並上傳至OSS。

    步驟五：建立Secret儲存OSS訪問憑據

    關於如何建立Secret用於儲存OSS訪問憑據，參見步驟三：建立Secret儲存OSS訪問憑據。

    步驟六：提交樣本Spark作業

    根據如下內容建立SparkApplication資訊清單檔並儲存為spark-pagerank.yaml。將<SPARK_IMAGE>替換為您在步驟三：構建Spark容器鏡像的倉庫地址，同時將<OSS_BUCKET>和<OSS_ENDPOINT>替換成您的OSS儲存桶名稱和訪問端點。關於如何在Spark作業中配置Celeborn的更多資訊，請參見Celeborn使用文檔。

    apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: <SPARK_IMAGE>                                     # Spark 鏡像，將<SPARK_IMAGE>替換成Spark鏡像名稱
  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           # 指定輸入測試資料集，將<OSS_BUCKET>替換成OSS Buckt名稱。
  - "10"                                                   # 迭代次數。
  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>                        # OSS訪問端點。例如北京地區OSS的內網訪問地址為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

    （可選）步驟七：環境清理

    如果您已體驗完本教程，相關資源如不再需要，可以通過執行以下命令進行刪除。

    執行如下命令刪除Spark作業。

    kubectl delete sparkapplication spark-pagerank

    執行如下命令刪除Secret資源。

    kubectl delete secret spark-oss-secret

    相關文檔