Spark BlockManager 在本地主机上运行

2024-01-11

我有一个简单的脚本文件,我试图在模仿教程的 Spark-Shell 中执行here https://spark.apache.org/examples.html

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

sc.stop();

val conf = new SparkConf().setAppName("MyApp").setMaster("mesos://zk://172.24.51.171:2181/mesos").set("spark.executor.uri", "hdfs://172.24.51.171:8020/spark-1.3.0-bin-hadoop2.4.tgz").set("spark.driver.host", "172.24.51.142")

val sc2 = new SparkContext(conf)

val file = sc2.textFile("hdfs://172.24.51.171:8020/input/pg4300.txt")

val errors = file.filter(line => line.contains("ERROR"))

errors.count()

我的namenode和mesos master位于172.24.51.171,我的IP地址是172.24.51.142。我将这些行保存到一个文件中,然后使用以下命令启动该文件:

/opt/spark-1.3.0-bin-hadoop2.4/bin/spark-shell -i WordCount.scala

我的远程执行器都因类似于以下的错误而死亡:

15/04/08 14:30:39 ERROR RetryingBlockFetcher: Exception while beginning fetch of 1 outstanding blocks 
java.io.IOException: Failed to connect to localhost/127.0.0.1:48554
    at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:191)
    at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:156)
    at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:78)
    at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:140)
    at org.apache.spark.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:120)
    at org.apache.spark.network.netty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:87)
    at org.apache.spark.network.BlockTransferService.fetchBlockSync(BlockTransferService.scala:89)
    at org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:594)
    at org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:592)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
    at org.apache.spark.storage.BlockManager.doGetRemote(BlockManager.scala:592)
    at org.apache.spark.storage.BlockManager.getRemoteBytes(BlockManager.scala:586)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.org$apache$spark$broadcast$TorrentBroadcast$$anonfun$$getRemote$1(TorrentBroadcast.scala:126)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$1.apply(TorrentBroadcast.scala:136)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$1.apply(TorrentBroadcast.scala:136)
    at scala.Option.orElse(Option.scala:257)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply$mcVI$sp(TorrentBroadcast.scala:136)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply(TorrentBroadcast.scala:119)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1.apply(TorrentBroadcast.scala:119)
    at scala.collection.immutable.List.foreach(List.scala:318)
    at org.apache.spark.broadcast.TorrentBroadcast.org$apache$spark$broadcast$TorrentBroadcast$$readBlocks(TorrentBroadcast.scala:119)
    at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$readBroadcastBlock$1.apply(TorrentBroadcast.scala:174)
    at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1152)
    at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:164)
    at org.apache.spark.broadcast.TorrentBroadcast._value$lzycompute(TorrentBroadcast.scala:64)
    at org.apache.spark.broadcast.TorrentBroadcast._value(TorrentBroadcast.scala:64)
    at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:87)
    at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:58)
    at org.apache.spark.scheduler.Task.run(Task.scala:64)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.ConnectException: Connection refused: localhost/127.0.0.1:48554
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739)
    at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:208)
    at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:287)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:528)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116)
    ... 1 more 

我运行 error.count() 命令后发生此故障。在我的 shell 中,创建新的 SparkContext 后,我​​看到了以下几行:

15/04/08 14:31:18 INFO NettyBlockTransferService: Server created on 48554
15/04/08 14:31:18 INFO BlockManagerMaster: Trying to register BlockManager
15/04/08 14:31:18 INFO BlockManagerMasterActor: Registering block manager localhost:48554 with 265.4 MB RAM, BlockManagerId(<driver>, localhost, 48554)
15/04/08 14:31:18 INFO BlockManagerMaster: Registered BlockManager

我猜发生的情况是 Spark 将 BlockManager 的地址记录为 localhost:48554,然后将其发送给所有尝试与其 localhost:48554 通信的执行器,而不是端口 48554 处的驱动程序 IP 地址。为什么spark使用localhost作为BlockManager的地址而不是spark.driver.host?

附加信息

  1. In the 火花配置 http://spark.apache.org/docs/1.2.1/configuration.html有spark.blockManager.port但没有spark.blockManager.host?只有一个spark.driver.host,你可以看到我在SparkConf中设置了它。

  2. 可能与此有关吉拉门票 https://issues.apache.org/jira/browse/SPARK-6420尽管这似乎是一个网络问题。我的网络配置了 DNS 就很好了。


您可以尝试在调用spark-shell时使用--master参数提供Spark Master地址(或添加到spark-defaults.conf中)。我遇到了类似的问题(请参阅我的帖子Spark Shell 监听本地主机而不是配置的 IP 地址 https://stackoverflow.com/questions/29810054/spark-shell-listens-on-localhost-instead-of-configured-ip-address)并且看起来当在 shell 中动态创建上下文时,BlockManager 会监听 localhost。

Logs:

  • 当使用原始上下文时(侦听主机名) BlockManagerInfo:在 ubuntu64server2:33301 上的内存中添加了broadcast_1_piece0

  • 创建新上下文时(在本地主机上侦听) BlockManagerInfo:在本地主机的内存中添加了broadcast_1_piece0:40235

我必须连接到 Cassandra 集群,并且能够通过在 Spark-defaults.conf 中提供 Spark.cassandra.connection.host 并在 Spark shell 中导入包 com.datastax.spark.connector._ 来查询它。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

Spark BlockManager 在本地主机上运行 的相关文章

随机推荐