将文件名追加到 RDD

Question

我有一个包含我的数据文件的文件夹。每个文件的大小约为 1 GB。

我需要的是RDD里面的文件名。

以下未按预期工作：

import glob
rdds = []
for filename in glob.iglob('/data/*'):
    rdd = sc.textFile(filename).map(lambda row: (filename, row))
    rdds.append(rdd)

allData = sc.union(rdds)

使用这个，filename总是最后读取文件的文件名

我也尝试过：

import glob
rdds = []
for filename in glob.iglob('/data/*'):
    def f(name=filename):
        return name
    rdd = sc.textFile(filename).map(lambda row: (f(), row))
    rdds.append(rdd)

allData = sc.union(rdds)

但这会产生错误：Broadcast can only be serialized in driver

sc.wholeTextFile() 不是一个选项，因为单个文件太大。

Answer 1

我猜你想要这样的东西：

import functools

def proc(f):
    return sc.textFile(f).map(lambda x: (f, x))

rdd = functools.reduce(
    lambda rdd1, rdd2: rdd1.union(rdd2),
    (proc(f) for f in glob.glob("/data/*")))

或 sc.union:

sc.union([proc(f) for f in glob.glob("/data/*")])

将文件名追加到 RDD

Append Filename to RDD

python

apache-spark

pyspark