如何解决pyspark中的pickle错误？

Question

我遍历文件来收集有关字典中的列和行中的值的信息。我有以下代码在本地工作：

def search_nulls(file_name): 
    separator = ',' 
    nulls_dict = {} 
    fp = open(file_name,'r') 
    null_cols = {} 
    lines = fp.readlines() 

    for n,line in enumerate(lines): 
     line = line.split(separator) 
     for m,data in enumerate(line): 
      data = data.strip('\n').strip('\r') 
      if str(m) not in null_cols: 
       null_cols[str(m)] = defaultdict(lambda: 0) 
      if len(data) <= 4: 
       null_cols[str(m)][str(data)] = null_cols[str(m)][str(data)] + 1 

    return null_cols 


files_to_process = ['tempfile.csv'] 
results = map(lambda file: search_nulls(file), files_to_process) 

上面的代码工作正常，没有火花。 我评论的最后两行以上，我尝试用火花，因为这是一些原型，将需要运行分布：

os.environ['SPARK_HOME'] = <path_to_spark_folder> 
conf = SparkConf().setAppName("search_files").setMaster('local') 

sc = SparkContext(conf=conf) 

objects = sc.parallelize(files_to_process) 
resulting_object = \ 
    objects.map(lambda file_object: find_nulls(file_object)) 

result = resulting_object.collect() 

使用的火花，不过，这将导致以下错误：

File "<path-to-spark>/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main 
    process() 
    File "<path-to-spark>/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process 
    serializer.dump_stream(func(split_index, iterator), outfile) 
    File "<path-to-spark>/python/lib/pyspark.zip/pyspark/serializers.py", line 267, in dump_stream 
    bytes = self.serializer.dumps(vs) 
    File "<path-to-spark>/python/lib/pyspark.zip/pyspark/serializers.py", line 415, in dumps 
    return pickle.dumps(obj, protocol) 
TypeError: expected string or Unicode object, NoneType found​ 

我一直无法找到任何明显的原因，为什么这会失败，因为它在本地完美运行，并且我没有在工作节点间共享任何文件。实际上，我只是在本地机器上运行它。

有谁知道为什么这可能会失败的一个很好的理由？

Answer 1

你的问题的根源是以下行：

null_cols[str(m)] = defaultdict(lambda: 0) 

正如你可以在the pickle module documentation的What can be pickled and unpickled?部分阅读：

The following types can be pickled:

• ...
• functions defined at the top level of a module (using def, not lambda)
• built-in functions defined at the top level of a module
• ...

应该明确的是，lambda: 0不符合以上标准。为了使它工作，你可以例如用int替换lambda表达式：

null_cols[str(m)] = defaultdict(int) 

这怎么可能，我们可以通过lambda表达式中PySpark高阶功能呢？魔鬼在细节中。 PySpark根据上下文使用不同的序列化器。要序列化闭包，包括lambda表达式，它使用定制的cloudpickle，它支持lambda表达式和嵌套函数。为了处理数据，它使用默认的Python工具。

---

一些旁注：

• 我不会使用Python file对象读取数据。它不可移植，不会超出本地文件系统。您可以改用SparkContex.wholeTextFiles。
• 如果确实关闭连接。使用with声明通常是最好的方法
• 你可以放心地剥离换行符拆分前行