PySpark PicklingError答案

【问题标题】：PySpark PicklingErrorPySpark PicklingError
【发布时间】：2016-12-12 21:33:09
【问题描述】：

我看到一个酸洗错误：

由于需要过深的递归，因此无法腌制对象。

以下是回溯：

Traceback (most recent call last):
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/streaming/util.py", line 62, in call
    r = self.func(t, *rdds)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/streaming/dstream.py", line 159, in 
    func = lambda t, rdd: old_func(rdd)
    if rdd.count() > 0:
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 1006, in count
    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 997, in sum
    return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 871, in fold
    vals = self.mapPartitions(func).collect()
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 773, in collect
    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2388, in _jrdd
    pickled_cmd, bvars, env, includes = _prepare_for_python_RDD(self.ctx, command, self)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2308, in _prepare_for_python_RDD
    pickled_command = ser.dumps(command)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 428, in dumps
    return cloudpickle.dumps(obj, 2)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 646, in dumps
    cp.dump(obj)
  File "/usr/hdp/current/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 111, in dump
    raise pickle.PicklingError(msg)
   PicklingError: Could not pickle object as excessively deep recursion required.

这是导致错误的代码 inn highlevel 的一部分：

sc = SparkContext(appName="my_app")

ssc = StreamingContext(sc, 1)

kafka_stream = KafkaUtils.createDirectStream(ssc, full_topic_list, kafka_params, fromOffsets=offset_dict)

messages = kafka_stream.map(lambda (k, v): json.loads(v))

messages.foreachRDD(lambda rdd: process(rdd, topic_list, sqlcontext))

在我的进程函数中，有一个 rdd 计数：if topic_rdd.count() > 0，它会抛出错误。

【问题讨论】：

标签： apache-kafka pyspark pickle rdd

【解决方案1】：

您不能将 RDD 传递给分布式函数（maps、reduce 等）并在其中处理 RDD。

【讨论】：

感谢 cftarnas。如果我通过 RDD，为 rdd 执行功能的最佳方法是什么。
@ling：你真的不能通过 RDD，它们不可腌制。如何解决您的整体问题取决于您要做什么，更详细的代码 sn-p 可能会有所帮助。它可以像预先计算 topic_rdd.count() 一样简单，然后自己传递计数。