在 Kubernetes 上运行 Spark 时工作人员出错

Question

我在 Kubernetes 上以客户端模式运行 Spark 2.4.1。

我正在尝试从包含 spark 的 pod 提交任务，该任务将启动 2 个执行器 pod。 命令如下：

bin/spark-shell \
--master k8s://https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT \
--deploy-mode client \
--conf spark.executor.instances=2 \
--conf spark.kubernetes.container.image=$SPARK_IMAGE \
--conf spark.kubernetes.driver.pod.name=$HOSTNAME
--conf spark.kubernetes.executor.podNamePrefix=spark-exec \
--conf spark.ui.port=4040

这些执行程序 pod 已创建，但一直失败并出现错误：

Caused by: java.io.IOException: Failed to connect to spark-57b8f99554-7nd45:4444
at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:245)
at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:187)
at org.apache.spark.rpc.netty.NettyRpcEnv.createClient(NettyRpcEnv.scala:198)
at org.apache.spark.rpc.netty.Outbox$$anon$1.call(Outbox.scala:194)
at org.apache.spark.rpc.netty.Outbox$$anon$1.call(Outbox.scala:190)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748) 
Caused by: java.net.UnknownHostException: spark-57b8f99554-7nd45
at java.net.InetAddress.getAllByName0(InetAddress.java:1281)
at java.net.InetAddress.getAllByName(InetAddress.java:1193)
at java.net.InetAddress.getAllByName(InetAddress.java:1127)
at java.net.InetAddress.getByName(InetAddress.java:1077)
at io.netty.util.internal.SocketUtils$8.run(SocketUtils.java:146)
at io.netty.util.internal.SocketUtils$8.run(SocketUtils.java:143)

似乎工作 pod 无法到达主节点（pod spark-57b8f99554-7nd45），它应该与 THIS 点有关，但我不知道如何解决它。 有什么想法吗？

Answer 1

要在 Kubernetes pod 上以客户端模式运行 Spark，您需要执行以下步骤：

1. 创建一个像这样的无头服务：

   apiVersion: v1
   kind: Service
   metadata:
     name: yoursparkapp
   spec:
     clusterIP: "None"
     selector:
       spark-app-selector: yoursparkapp
   ports:
     - name: driver-rpc-port
       protocol: TCP 
       port: 7078
       targetPort: 7078
     - name: blockmanager
       protocol: TCP 
       port: 7079
       targetPort: 7079
   

注意这部分：spark-app-selector: yoursparkapp，因为它必须与用于运行将执行 spark-submit 的 pod 的标签匹配。

使用以下命令在您的集群中安装上述服务：kubectl create -f yoursparkappservice.yml -n your_namespace

1. 运行一些分配上述服务的 pod：

   kubectl run  \
   -n your_namespace -i --tty yoursparkapp \
   --restart=Never \
   --overrides='{ "apiVersion" : "v1", "metadata" : { "annotations" : "labels": { "spark-app-selector" : "yoursparkapp" } } }' \
   --image=your_container:latest -- /bin/bash
   

对于标签，我们使用"spark-app-selector" : "yoursparkapp"。这样，这个 pod 就会使用第一步创建的服务。

1. 在第 2 步创建的 pod 中，我们可以执行 spark-submit：

   spark-submit --master k8s://https://kubernetes_url:443 \
   --deploy-mode client \
   --name yoursparkapp \
   --conf spark.kubernetes.container.image=your_container:latest \
   --conf spark.kubernetes.pyspark.pythonVersion=3 \
   --conf spark.kubernetes.namespace=your_namespace \
   --conf spark.kubernetes.container.image.pullPolicy=Always \
   --conf spark.driver.memory=2g  \
   --conf spark.executor.memory=2g \
   --conf spark.submit.deployMode=client \
   --conf spark.executor.cores=3 \
   --conf spark.driver.cores=3 \
   --conf spark.driver.host=yoursparkapp \
   --conf spark.driver.port=7078 \
   --conf spark.kubernetes.driver.pod.name=yoursparkapp  \
    /path/to/your/remote_spark_app.py
   

Answer 2

为了从 Kubernetes pod 提交任务，您应该在 spark-submit 中定义这些参数：

--conf spark.driver.bindAddress=0.0.0.0
--conf spark.driver.host=$MY_NODE_NAME
--conf spark.driver.port=3xxx1
--conf spark.driver.blockManager.port=3xxx2

您可以像这样在服务和部署配置中定义端口和节点名称（我省略了一般部分）：

kind: Service
spec:
  type: NodePort
  ports:
  - name: port-1
    nodePort: 3xxx1
    port: 3xxx1
    protocol: TCP
    targetPort: 3xxx1
  - name: port-2
    nodePort: 3xxx2
    port: 3xxx2
    protocol: TCP
---
kind: Deployment
spec:
  template:
    spec:
      containers:
      - env:
          - name: MY_NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName