apache-spark - 在 Spark 中处理 bzipped json 文件？

Question

我在S3中有大约200个文件，例如a_file.json.bz2，这些文件的每一行都是JSON格式的记录，但有些字段是由pickle.dumps，例如一个日期时间字段。 bzip 压缩后每个文件约 1GB。现在我需要在 Spark(实际上是 pyspark)中处理这些文件，但我什至无法获取每条记录。那么这里的最佳实践是什么？

ds.take(10) 给出

[(0, u'(I551'),
 (6, u'(dp0'),
 (11, u'Vadv_id'),
 (19, u'p1'),
 (22, u'V479883'),
 (30, u'p2'),
 (33, u'sVcpg_id'),
 (42, u'p3'),
 (45, u'V1913398'),
 (54, u'p4')]

显然，拆分不是按每个记录进行的。

谢谢。

Answer 1

我遇到了这个问题 reading gpg-encrypted files 。您可以按照 Daniel 的建议使用 WholeTextFiles，但在读取大文件时必须小心，因为整个文件将在处理之前加载到内存中。如果文件太大，可能会使执行器崩溃。我使用了parallelize和flatMap。也许类似于

def read_fun_generator(filename):
    with bz2.open(filename, 'rb') as f:
        for line in f:
            yield line.strip()

bz2_filelist = glob.glob("/path/to/files/*.bz2")
rdd_from_bz2 = sc.parallelize(bz2_filelist).flatMap(read_fun_generator)