PySpark Standalone：java.lang.IllegalStateException：未读块数据

Question

我对使用 pyspark 还很陌生，我一直在尝试运行一个在本地模式下使用 1000 行数据子集运行良好的脚本，但现在在独立模式下抛出错误包含所有数据，即 1GB。我认为这会随着更多数据 = 更多问题而发生，但我无法理解导致此问题的原因。这些是我的独立集群的详细信息：

3 位执行者
每个 20GB 内存
spark.driver.maxResultSize=1GB（添加这个 bc 我认为这可能是问题所在，但没有解决问题）

脚本在我将 spark 数据帧转换为 pandas 数据帧以并行化某些操作的阶段抛出错误。我很困惑这会导致问题，因为数据只有大约 1G，而我的执行者应该有比这更多的内存。这是我的代码片段 - 错误发生在 data = data.toPandas():

def num_cruncher(data, cols=[], target='RETAINED', lvl='univariate'):
    if not cols:
            cols = data.columns
            del cols[data.columns.index(target)]
    data = data.toPandas()
    pop_mean = data.mean()[0]
    if lvl=='univariate':
        cols = sc.parallelize(cols)
        all_df = cols.map(lambda x: calculate([x], data, target)).collect()
    elif lvl=='bivariate':
        cols = sc.parallelize(cols)
        cols = cols.cartesian(cols).filter(lambda x: x[0]<x[1])
        all_df = cols.map(lambda x: calculate(list(x), data, target)).collect()
    elif lvl=='trivariate':
        cols = sc.parallelize(cols)
        cols = cols.cartesian(cols).cartesian(cols).filter(lambda x: x[0][0]<x[0][1] and x[0][0]<x[1] and x[0][1]<x[1]).map(lambda x: (x[0][0],x[0][1],x[1]))
        all_df = cols.map(lambda x: calculate(list(x), data, target)).collect()
    all_df = pd.concat(all_df)
    return all_df, pop_mean

这是错误日志：

    16/07/11 09:49:54 ERROR TransportRequestHandler: Error while invoking RpcHandler#receive() for one-way message.
java.lang.IllegalStateException: unread block data
    at java.io.ObjectInputStream$BlockDataInputStream.setBlockDataMode(ObjectInputStream.java:2424)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1383)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
    at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:109)
    at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$deserialize$$anonfun$apply.apply(NettyRpcEnv.scala:258)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
    at org.apache.spark.rpc.netty.NettyRpcEnv.deserialize(NettyRpcEnv.scala:310)
    at org.apache.spark.rpc.netty.NettyRpcEnv$$anonfun$deserialize.apply(NettyRpcEnv.scala:257)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
    at org.apache.spark.rpc.netty.NettyRpcEnv.deserialize(NettyRpcEnv.scala:256)
    at org.apache.spark.rpc.netty.NettyRpcHandler.internalReceive(NettyRpcEnv.scala:588)
    at org.apache.spark.rpc.netty.NettyRpcHandler.receive(NettyRpcEnv.scala:577)
    at org.apache.spark.network.server.TransportRequestHandler.processOneWayMessage(TransportRequestHandler.java:170)
    at org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:104)
    at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:104)
    at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:51)
    at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:266)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at org.apache.spark.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:86)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    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.run(SingleThreadEventExecutor.java:111)
    at java.lang.Thread.run(Thread.java:745)

所以我的问题是：

为什么给工作人员 20GB 的内存对于这个 1GB 的数据集来说还不够？
一般来说，像我在这里做的那样将数据加载到内存中是个好主意还是有更好的方法来做到这一点？

Answer 1

对于可能发现此 post 有用的人 - 问题似乎不是为 worker/slaves 提供更多内存，而是为驱动程序提供更多内存，如@KartikKannapur 的评论。所以为了解决这个问题，我设置了：

spark.driver.maxResultSize 3g
spark.driver.memory 8g
spark.executor.memory 4g

可能有点矫枉过正，但现在可以了。

PySpark Standalone：java.lang.IllegalStateException：未读块数据

PySpark Standalone: java.lang.IllegalStateException: unread block data

apache-spark

pyspark

spark-dataframe