在Docker容器中运行Spark驱动程序 - 执行程序没有回连到驱动程序?

8

更新:问题已解决。Docker镜像在此:docker-spark-submit

我在Docker容器中运行带有fat jar的spark-submit。我的独立Spark集群在3个虚拟机上运行,其中一个是master,另外两个是worker。从worker机器上的执行者日志中,我看到执行者具有以下驱动程序URL:

"--driver-url" "spark://CoarseGrainedScheduler@172.17.0.2:5001"

172.17.0.2实际上是驱动程序所在容器的地址,而不是容器运行的主机地址。这个IP地址无法从worker机器上访问,因此worker无法与驱动程序通信。 从StandaloneSchedulerBackend的源代码中可以看出,它使用spark.driver.host设置构建driverUrl:

val driverUrl = RpcEndpointAddress(
  sc.conf.get("spark.driver.host"),
  sc.conf.get("spark.driver.port").toInt,
  CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString

它没有考虑到SPARK_PUBLIC_DNS环境变量 - 这是正确的吗?在容器中,我不能将spark.driver.host设置为除容器“内部”IP地址(例如本例中的172.17.0.2)以外的任何其他值。尝试将spark.driver.host设置为主机机器的IP地址时,会出现如下错误: WARN Utils:服务'sparkDriver'无法绑定到端口5001。尝试端口5002。
我尝试将spark.driver.bindAddress设置为主机机器的IP地址,但仍然遇到相同的错误。那么,如何配置Spark使用主机机器IP地址而不是Docker容器地址与驱动程序进行通信?
更新:来自执行程序的堆栈跟踪:
ERROR RpcOutboxMessage: Ask timeout before connecting successfully
Exception in thread "main" java.lang.reflect.UndeclaredThrowableException
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1713)
    at org.apache.spark.deploy.SparkHadoopUtil.runAsSparkUser(SparkHadoopUtil.scala:66)
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.run(CoarseGrainedExecutorBackend.scala:188)
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.main(CoarseGrainedExecutorBackend.scala:284)
    at org.apache.spark.executor.CoarseGrainedExecutorBackend.main(CoarseGrainedExecutorBackend.scala)
Caused by: org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 120 seconds. This timeout is controlled by spark.rpc.askTimeout
    at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
    at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
    at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216)
    at scala.util.Try$.apply(Try.scala:192)
    at scala.util.Failure.recover(Try.scala:216)
    at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:326)
    at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:326)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
    at org.spark_project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293)
    at scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
    at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
    at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
    at scala.concurrent.Promise$class.complete(Promise.scala:55)
    at scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153)
    at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237)
    at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
    at scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:63)
    at scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:78)
    at scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
    at scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
    at scala.concurrent.BatchingExecutor$Batch.run(BatchingExecutor.scala:54)
    at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
    at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:106)
    at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
    at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
    at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
    at scala.concurrent.Promise$class.tryFailure(Promise.scala:112)
    at scala.concurrent.impl.Promise$DefaultPromise.tryFailure(Promise.scala:153)
    at org.apache.spark.rpc.netty.NettyRpcEnv.org$apache$spark$rpc$netty$NettyRpcEnv$$onFailure$1(NettyRpcEnv.scala:205)
    at org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:239)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    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.util.concurrent.TimeoutException: Cannot receive any reply in 120 seconds
    ... 8 more

以下答案非常好,解释得很清楚。另一种方法是在独立模式下使用集群部署模式。这是您参考的链接:https://spark.apache.org/docs/latest/submitting-applications.html - Daisy QL
3个回答

7

因此,有效的配置如下:

  • 将spark.driver.host设置为主机的IP地址
  • 将spark.driver.bindAddress设置为容器的IP地址

有效的Docker镜像在这里:docker-spark-submit


当使用SPARK_DRIVER_HOST时它没有起作用。显然,在这里有三个IP地址:docker0:172.17.0.1(主机的Docker以太网),作业服务器的Docker IP(172.17.0.2)和主机本地分配的IP地址。我们尝试将这三个IP地址都用作SPARK_DRIVER_HOST,但是我们仍然得到连接超时的错误!! - Somum
@Somum 上述解决方案适用于Apache Spark-2.1.0+。请检查您的Spark版本。 - Sangram Gaikwad
bindAddress should probably be set to 0.0.0.0 - OneCricketeer

6

我注意到其他答案使用的是Spark Standalone(如OP所提到的VM或其他答案中的127.0.0.1)。

我想展示一下我运行变体的jupyter/pyspark-notebook,连接到远程AWS Mesos集群,在本地Mac上在Docker中运行容器时似乎有效。

在这种情况下,这些说明适用,但是--net=host只能在Linux主机上工作。
这里有一个重要的步骤 - 按照链接中提到的在Mesos从节点的操作系统上创建笔记本用户。

这个图表对于调试网络很有帮助,但它没有提到spark.driver.blockManager.port,实际上这是最终使其工作的参数,我在Spark文档中错过了这个参数。否则,Mesos从节点上的执行程序也会尝试绑定该块管理器端口,并且Mesos会拒绝分配它。

enter image description here

公开这些端口以便您可以本地访问Jupyter和Spark UI

  • Jupyter UI (8888)
  • Spark UI (4040)

并且这些端口使得Mesos可以回到驱动程序:重要提示: Mesos主节点、从节点和Zookepeeper也必须允许双向通信...

  • "libprocess"地址+端口似乎通过LIBPROCESS_PORT变量(随机:37899)存储/广播到Zookeeper中。参见:Mesos文档
  • Spark driver端口(随机:33139)+ 16用于spark.port.maxRetries
  • Spark块管理器端口(随机:45029)+ 16用于spark.port.maxRetries

不是特别相关,但我正在使用Jupyter Lab界面

export EXT_IP=<your external IP>

docker run \
  -p 8888:8888 -p 4040:4040 \
  -p 37899:37899 \
  -p 33139-33155:33139-33155 \
  -p 45029-45045:45029-45045 \
  -e JUPYTER_ENABLE_LAB=y \
  -e EXT_IP \
  -e LIBPROCESS_ADVERTISE_IP=${EXT_IP} \
  -e LIBPROCESS_PORT=37899 \
  jupyter/pyspark-notebook

一旦开始,我就会打开 Jupyter 的地址 localhost:8888,并为简单的 spark-shell 操作打开一个终端。我也可以添加一个卷挂载来运行实际的打包代码,但那是下一步。我没有编辑 spark-env.shspark-default.conf,所以现在我只需要将所有相关的配置传递给 spark-shell。提醒:这是在容器内操作的。
spark-shell --master mesos://zk://quorum.in.aws:2181/mesos \
  --conf spark.executor.uri=https://path.to.http.server/spark-2.4.2-bin-hadoop2.7.tgz \
  --conf spark.cores.max=1 \
  --conf spark.executor.memory=1024m \
  --conf spark.driver.host=$LIBPROCESS_ADVERTISE_IP \
  --conf spark.driver.bindAddress=0.0.0.0 \
  --conf spark.driver.port=33139 \
  --conf spark.driver.blockManager.port=45029

这将加载Spark REPL,在一些有关查找Mesos主节点并注册框架的输出之后,我随后使用NameNode IP从HDFS中读取了一些文件(尽管我认为任何其他可访问的文件系统或数据库应该也可以工作)

然后我得到了预期的输出

Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.2
      /_/

Using Scala version 2.12.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_202)
Type in expressions to have them evaluated.
Type :help for more information.

scala> spark.read.text("hdfs://some.hdfs.namenode:9000/tmp/README.md").show(10)
+--------------------+
|               value|
+--------------------+
|      # Apache Spark|
|                    |
|Spark is a fast a...|
|high-level APIs i...|
|supports general ...|
|rich set of highe...|
|MLlib for machine...|
|and Spark Streami...|
|                    |
|<http://spark.apa...|
+--------------------+
only showing top 10 rows

1
我在我的Jupyter容器中使用了相同的镜像。打开端口是关键。 - Daisy QL

5

我的设置,使用 Docker 和 MacOS:

  • 在同一个 Docker 容器中运行 Spark 1.6.3 主节点和工作节点
  • 从 MacOS(通过 IDE)运行 Java 应用程序

Docker-compose 打开端口:

ports:
- 7077:7077
- 20002:20002
- 6060:6060

Java配置(用于开发目的):

        esSparkConf.setMaster("spark://127.0.0.1:7077");
        esSparkConf.setAppName("datahub_dev");

        esSparkConf.setIfMissing("spark.driver.port", "20002");
        esSparkConf.setIfMissing("spark.driver.host", "MAC_OS_LAN_IP");
        esSparkConf.setIfMissing("spark.driver.bindAddress", "0.0.0.0");
        esSparkConf.setIfMissing("spark.blockManager.port", "6060");

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接