pyspark 流式传输：无法对工作人员执行 rdd.count()答案

【问题标题】：pyspark streaming: failing to execute rdd.count() on workerspyspark 流式传输：无法对工作人员执行 rdd.count()
【发布时间】：2018-07-12 17:42:02
【问题描述】：

我有一个 pyspark 流式传输工作，按照这些思路做一些事情：

def printrddcount(rdd):
    c = rdd.count()
    print("{1}: Received an RDD of {0} rows".format("CANNOTCOUNT", datetime.now().isoformat()) )

然后：

...
stream.foreachRDD(printrddcount)

据我所知，printrdd 函数将在工作人员中执行而且，是的，我知道在工作人员中执行 print() 是个坏主意。但这不是重点。我很确定这段代码直到最近才有效。（而且，它看起来不一样，因为 'c' 的内容实际上是打印在 print 语句中，而不是仅仅收集，然后丢弃......）

但现在，似乎（突然？），然后 rdd.count() 已经停止工作，并且正在让我的工作进程死亡说：

UnpicklingError: NEWOBJ class argument has NULL tp_new

完整的（好吧，仅限 python）堆栈跟踪：

Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/worker.py", line 163, in main
    func, profiler, deserializer, serializer = read_command(pickleSer, infile)
  File "/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/worker.py", line 54, in read_command
command = serializer._read_with_length(file)
  File "/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/serializers.py", line 169, in _read_with_length
    return self.loads(obj)
  File "/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/serializers.py", line 454, in loads
    return pickle.loads(obj)
UnpicklingError: NEWOBJ class argument has NULL tp_new

确实，它失败的那一行是 rdd.count()

知道为什么 rdd.count() 会失败吗？如果某些东西应该被序列化，它应该是 rdd，对吧？

【问题讨论】：

标签： apache-spark pyspark spark-streaming

【解决方案1】：

好的。我进一步调查了一下。 rdd.count() 没什么问题

唯一的错误是管道上有另一个转换以某种方式“损坏”（关闭？无效？沿着这些线的东西）rdd。因此，当它到达 printrddcount 函数时，它不能再被序列化并给出错误。

问题出在如下代码中：

...
log = logging.getLogger(__name__)
...
def parse(parse_function):
    def parse_function_wrapper(event):
        try:
            log.info("parsing")
            new_event = parse_function(event)
    except ParsingFailedException as e:
        pass
    return new_event
return parse_function_wrapper

然后：

stream = stream.map(parse(parse_event))

现在，log.info（尝试了很多变体，一开始日志记录在异常处理程序中）是造成问题的原因。这让我说，最有可能的是，由于某种原因，它是无法序列化的记录器对象。

自己关闭这个线程，因为它实际上与 rdd 序列化无关；甚至很可能连 pyspark 都没有。

【讨论】：