python - 如何在 docker 中从 python 连接到远程 Spark 集群

Question

我在用户为 docker-user 的容器中安装了 Spark 2.0.0 和 Python 3。独立模式似乎有效。

我们已经在 AWS 和 hadoop 上搭建了一个 Spark 集群。在运行 VPN 的情况下，我可以从笔记本电脑通过 ssh 连接到“内部 IP”，例如

ssh ubuntu@1.1.1.1

这就登录了，然后

cd /opt/spark/bin
./pyspark

这显示了 Spark 2.0.0 和 Python 2.7.6。一个天真的 parallelize 示例可以工作。

现在在 Docker 支持的 Jupyter Notebook 中，执行

from pyspark import SparkConf, SparkContext
conf = SparkConf().setAppName('hello').setMaster('spark://1.1.1.1:7077').setSparkHome('/opt/spark/')
sc = SparkContext(conf=conf)

这显然进入了集群，因为我可以在 1.1.1.1:8080 的 Spark 仪表板中看到应用程序“hello”。令我感到困惑的是，它在 Docker 内部已经走了这么远，而不关心 ssh、密码等。

现在尝试一个简单的 parallelize 示例，

x = ['spark', 'rdd', 'example', 'sample', 'example']
y = sc.parallelize(x)

看起来不错。然后，

y.collect()

它卡在那里。

在仪表板“Executor Summary”表上，我不知道要查找什么。但是一个状态为 exited 的 worker 有这样的 stderr:

16/08/16 17:37:01 INFO SignalUtils: Registered signal handler for TERM
16/08/16 17:37:01 INFO SignalUtils: Registered signal handler for HUP
16/08/16 17:37:01 INFO SignalUtils: Registered signal handler for INT
16/08/16 17:37:02 INFO SecurityManager: Changing view acls to: ubuntu,docker-user
16/08/16 17:37:02 INFO SecurityManager: Changing modify acls to: ubuntu,docker-user
16/08/16 17:37:02 INFO SecurityManager: Changing view acls groups to: 
16/08/16 17:37:02 INFO SecurityManager: Changing modify acls groups to: 
16/08/16 17:37:02 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(ubuntu, docker-user); groups with view permissions: Set(); users  with modify permissions: Set(ubuntu, docker-user); groups with modify permissions: Set()
Exception in thread "main" java.lang.reflect.UndeclaredThrowableException
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1671)
    at org.apache.spark.deploy.SparkHadoopUtil.runAsSparkUser(SparkHadoopUtil.scala:70)
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.run(CoarseGrainedExecutorBackend.scala:166)
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.main(CoarseGrainedExecutorBackend.scala:262)
    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:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.TimeoutException: Cannot receive any reply in 120 seconds
    ... 8 more
java.lang.IllegalArgumentException: requirement failed: TransportClient has not yet been set.
    at scala.Predef$.require(Predef.scala:224)
    at org.apache.spark.rpc.netty.RpcOutboxMessage.onTimeout(Outbox.scala:70)
    at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$ask$1.applyOrElse(NettyRpcEnv.scala:232)
    at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$ask$1.applyOrElse(NettyRpcEnv.scala:231)
    at scala.concurrent.Future$$anonfun$onFailure$1.apply(Future.scala:138)
    at scala.concurrent.Future$$anonfun$onFailure$1.apply(Future.scala:136)
    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.tryFailure(Promise.scala:112)

注意 Docker 用户 docker-user 可能是个问题，因为那里的服务器机器需要 ubuntu。可能还有其他问题。

Python 包 paramiko 能帮上忙吗？我知道如何使用 paramiko 创建一个客户端对象，通过它发出命令等，就像我登录到服务器一样。但是不知道如何将它与 SparkConf 和 SparkContext 结合起来。

各种来源停止说 SparkConf().setMaster('spark://1.1.1.1:7077') 就好像它会起作用一样。我相信关于登录、密码、ssh、auth 的一些麻烦是不可避免的。

谢谢!

Answer 1

spark 驱动程序必须可以从集群访问，确保您可以 ping 运行 spark 驱动程序的机器。这是因为执行者必须主动联系司机。它们不会保持 TCP 连接处于事件状态(否则无法扩展)。

另一种方法是使用集群模式而不是客户端模式。