运行 Docker 容器中的 Spark 驱动程序 - 没有从执行程序返回到驱动程序的连接？
Running Spark driver program in Docker container - no connection back from executor to the driver?
更新：问题已解决。 Docker 图片在这里：docker-spark-submit
我 运行 spark-submit 在 Docker 容器中有一个 fat jar。我的独立 Spark 集群 运行s 在 3 个虚拟机上 - 一个主节点和两个工作节点。从工作机器上的执行程序日志中，我看到执行程序具有以下驱动程序 URL：
"--driver-url" "spark://CoarseGrainedScheduler@172.17.0.2:5001"
172.17.0.2其实是带有驱动程序的容器的地址，而不是运行ning容器所在的宿主机。 worker 机器无法访问此 IP，因此 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 环境变量 - 这是正确的吗？在容器中，除了容器 "internal" IP 地址（本例中为 172.17.0.2）之外，我无法将 spark.driver.host 设置为任何其他内容。尝试将 spark.driver.host 设置为主机的 IP 地址时，出现如下错误：
WARN Utils: Service 'sparkDriver' could not bind on port 5001.
Attempting port 5002.
我试图将 spark.driver.bindAddress 设置为主机的 IP 地址，但得到了同样的错误。
那么，如何配置 Spark 使用主机 IP 地址而不是 Docker 容器地址与驱动程序通信？
UPD：来自执行程序的堆栈跟踪：
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.applyOrElse(RpcTimeout.scala:63)
at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout.applyOrElse(RpcTimeout.scala:59)
at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
at scala.util.Failure$$anonfun$recover.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.apply(Future.scala:326)
at scala.concurrent.Future$$anonfun$recover.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.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.apply(Future.scala:237)
at scala.concurrent.Future$$anonfun$map.apply(Future.scala:237)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.processBatch(BatchingExecutor.scala:63)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.apply$mcV$sp(BatchingExecutor.scala:78)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.apply(BatchingExecutor.scala:55)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.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(NettyRpcEnv.scala:205)
at org.apache.spark.rpc.netty.NettyRpcEnv$$anon.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.access1(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
所以工作配置是：
- 设置spark.driver.host为主机的IP地址
- 设置spark.driver.bindAddress为容器的IP地址
工作 Docker 图片在这里：docker-spark-submit。
我的设置，Docker 和 MacOS：
- 运行 同一 Docker 容器内的 Spark 1.6.3 master + worker
- 运行 Java 来自 MacOS 的应用程序（通过 IDE）
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");
我注意到其他答案使用的是 Spark Standalone（在 VM 上，如 OP 或 127.0.0.1 提到的其他答案）。
我想展示什么似乎对我有用 运行 jupyter/pyspark-notebook 针对远程 AWS Mesos 集群的变体，运行 Docker 中的容器在 Mac 本地。
在这种情况下 these instuctions apply，然而，--net=host 除了在 Linux 主机上无法工作。
这里的重要步骤 - 在 Mesos 从属 OS 上创建笔记本用户，如 link 中所述。
This diagram 对调试网络很有帮助，但它没有提到 spark.driver.blockManager.port，这实际上是使它起作用的最后一个参数，它我错过了 Spark 文档。否则，Mesos slaves 上的执行者也会尝试绑定该块管理器端口，而 Mesos 拒绝分配它。
公开这些端口，以便您可以在本地访问 Jupyter 和 Spark UI
- Jupyter UI (
8888)
- 火花UI (
4040)
以及这些端口，以便 Mesos 可以返回到驱动程序：重要：Bi-directional 必须允许 Mesos Masters、Slaves 和 Zookepeeper 进行通信...
- "libprocess" 地址 + 端口似乎通过
LIBPROCESS_PORT 变量（随机：37899）在 Zookeeper 中获得 stored/broadcast。参考：Mesos documentation
- Spark 驱动程序端口（随机：33139）+ 16 for
spark.port.maxRetries
- Spark 块管理器端口（随机：45029）+ 16 for
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.sh 或 spark-default.conf，所以我暂时将所有相关的 confs 传递给 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 master 和注册框架的输出之后，我然后使用 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
更新：问题已解决。 Docker 图片在这里：docker-spark-submit
我 运行 spark-submit 在 Docker 容器中有一个 fat jar。我的独立 Spark 集群 运行s 在 3 个虚拟机上 - 一个主节点和两个工作节点。从工作机器上的执行程序日志中，我看到执行程序具有以下驱动程序 URL：
"--driver-url" "spark://CoarseGrainedScheduler@172.17.0.2:5001"
172.17.0.2其实是带有驱动程序的容器的地址，而不是运行ning容器所在的宿主机。 worker 机器无法访问此 IP，因此 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 环境变量 - 这是正确的吗？在容器中，除了容器 "internal" IP 地址（本例中为 172.17.0.2）之外，我无法将 spark.driver.host 设置为任何其他内容。尝试将 spark.driver.host 设置为主机的 IP 地址时，出现如下错误：
WARN Utils: Service 'sparkDriver' could not bind on port 5001. Attempting port 5002.
我试图将 spark.driver.bindAddress 设置为主机的 IP 地址，但得到了同样的错误。 那么，如何配置 Spark 使用主机 IP 地址而不是 Docker 容器地址与驱动程序通信？
UPD：来自执行程序的堆栈跟踪：
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.applyOrElse(RpcTimeout.scala:63)
at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout.applyOrElse(RpcTimeout.scala:59)
at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
at scala.util.Failure$$anonfun$recover.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.apply(Future.scala:326)
at scala.concurrent.Future$$anonfun$recover.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.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.apply(Future.scala:237)
at scala.concurrent.Future$$anonfun$map.apply(Future.scala:237)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.processBatch(BatchingExecutor.scala:63)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.apply$mcV$sp(BatchingExecutor.scala:78)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.apply(BatchingExecutor.scala:55)
at scala.concurrent.BatchingExecutor$Batch$$anonfun$run.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(NettyRpcEnv.scala:205)
at org.apache.spark.rpc.netty.NettyRpcEnv$$anon.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.access1(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
所以工作配置是：
- 设置spark.driver.host为主机的IP地址
- 设置spark.driver.bindAddress为容器的IP地址
工作 Docker 图片在这里：docker-spark-submit。
我的设置，Docker 和 MacOS：
- 运行 同一 Docker 容器内的 Spark 1.6.3 master + worker
- 运行 Java 来自 MacOS 的应用程序（通过 IDE）
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");
我注意到其他答案使用的是 Spark Standalone（在 VM 上，如 OP 或 127.0.0.1 提到的其他答案）。
我想展示什么似乎对我有用 运行 jupyter/pyspark-notebook 针对远程 AWS Mesos 集群的变体，运行 Docker 中的容器在 Mac 本地。
在这种情况下 these instuctions apply，然而，--net=host 除了在 Linux 主机上无法工作。
这里的重要步骤 - 在 Mesos 从属 OS 上创建笔记本用户，如 link 中所述。
This diagram 对调试网络很有帮助，但它没有提到 spark.driver.blockManager.port，这实际上是使它起作用的最后一个参数，它我错过了 Spark 文档。否则，Mesos slaves 上的执行者也会尝试绑定该块管理器端口，而 Mesos 拒绝分配它。
公开这些端口，以便您可以在本地访问 Jupyter 和 Spark UI
- Jupyter UI (
8888) - 火花UI (
4040)
以及这些端口，以便 Mesos 可以返回到驱动程序：重要：Bi-directional 必须允许 Mesos Masters、Slaves 和 Zookepeeper 进行通信...
- "libprocess" 地址 + 端口似乎通过
LIBPROCESS_PORT变量（随机：37899）在 Zookeeper 中获得 stored/broadcast。参考：Mesos documentation - Spark 驱动程序端口（随机：33139）+ 16 for
spark.port.maxRetries - Spark 块管理器端口（随机：45029）+ 16 for
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.sh 或 spark-default.conf，所以我暂时将所有相关的 confs 传递给 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 master 和注册框架的输出之后，我然后使用 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