python之使用 PySpark 从 map 创建全局列表的问题

我有这段代码，我正在使用 pyspark 在 ipython 中读取文件。我想要做的是向其中添加一个片段，该片段基于从文件中读取的特定列形成一个列表，但是当我尝试执行它时，列表结果为空，并且没有任何内容附加到它。我的代码是:

list1 = [] 
 
def file_read(line): 
 
    list1.append(line[10]) 
    # bunch of other code which process other column indexes on `line` 
 
inputData = sc.textFile(fileName).zipWithIndex().filter(lambda (line,rownum): rownum>0).map(lambda (line, rownum): line) 
 
column_val = (inputData 
    .map(lambda line: line.split(",")) 
    .filter(lambda line: len(line) >1 ) 
    .map(file_read)) 

当我执行这部分代码时，list1 仍然是空的，即使 line[10] 中有数据，因为我在其他部分使用它上面相同函数中的代码。似乎它只是没有将它附加到列表中。我怎样才能形成上面的列表？

请您参考如下方法:

好吧，它实际上确实追加到 list1，问题不在于您正在考虑的那个。闭包中引用的每个变量都被序列化并发送给工作人员。它也适用于 list1。

每个分区都会收到它自己的 list1 副本，当 file_read 被调用时，数据会附加到该副本，并且当给定的映射阶段完成时，它会退出范围并被丢弃。

不是特别优雅的一段代码，但你应该看到这里确实发生了什么:

rdd = sc.parallelize(range(100), 5) 
 
line1 = [] 
 
def file_read(line): 
    list1.append(line) 
    print len(list1) 
    return line 
 
xs = rdd.map(file_read).collect() 

编辑

Spark 提供了两种类型的共享变量。 Broadcast variables ，从 worker 的角度来看是只读的，和accumulators仅从驱动程序的角度写入。

默认情况下，累加器仅支持数字变量，主要用作计数器。不过可以定义自定义累加器。为此，您必须扩展 AccumulatorParam类并提供定制zero和 addInPlace实现:

class ListParam(AccumulatorParam): 
    def zero(self, v): 
        return [] 
    def addInPlace(self, acc1, acc2): 
        acc1.extend(acc2) 
        return acc1 

接下来可以重新定义file_read如下:

def file_read1(line): 
    global list1 # Required otherwise the next line will fail 
    list1 += [line] 
    return line 

示例用法:

list1 = sc.accumulator([], ListParam()) 
 
rdd = sc.parallelize(range(10)).map(file_read1).collect() 
list1.value 

即使可以像这样使用累加器，但在实践中使用起来可能成本太高，而且在最坏的情况下，它可能会使驱动程序崩溃。相反，您可以简单地使用另一个转换:

tmp = (inputData 
    .map(lambda line: line.split(",")) 
    .filter(lambda line: len(line) >1 )) 
 
def line_read2(line): return ... # Just a core logic 
 
line1 = tmp.map(lambda line: line[10]) 
column_val = tmp.map(line_read2) 

旁注:

您提供的代码没有任何作用。 Spark 中的转换只是对必须完成的操作的描述，但在您调用操作数据之前，什么都不会真正执行。