hadoop - ArrayWritable 的 Pyspark 用法

Question

我尝试在 pyspark 上保存一个键值 RDD。 RDD 的每个单元都有类型，用以下伪代码编写:

((str,str),(int,[(int,int),...]))`

我想将它保存在 hadoop 文件系统上。为此，我将列表转换为元组并调用 .saveAsSequenceFile .但是用 sc.sequenceFile 重新加载 rdd失败 java.lang.RuntimeException: java.lang.NoSuchMethodException: org.apache.hadoop.io.ArrayWritable.<init>() .

这是一个试图保存 (int,int) 的 RDD 的最小示例.

import pyspark as spark, math

scConf = spark.SparkConf().setAppName('minimal_example')
sc = spark.SparkContext( conf = scConf )

def divs( x ):
    for n in xrange(1, int(math.sqrt(x))+1 ):
        if x % n == 0: yield n
def constructor( i ):
    return ( i, tuple(divs(i)) )

rdd = sc.parallelize(map(constructor,range(2,61)))
rdd.saveAsSequenceFile("min.seq")

当我用交互​​式 pyspark 加载它时使用 sc.sequenceFile，它也失败了。怎么了？为什么它在 python 中尝试保存数组，而我实际上有元组。以及如何在 pyspark 中扩展 ArrayWritable 以具有默认构造函数？

Answer 1

你真的需要序列文件吗？您可以保存为文本文件、加载文本文件和映射以恢复值。

rdd.saveAsTextFile('test') 

sc.textFile('test').map(lambda row: ast.literal_eval(row))