json - PySpark:如何读取多个 JSON 文件，每个文件有多条记录

Question

我有一个存储在 S3 存储桶中的大型数据集，但它不是单个大文件，而是由许多(准确地说是 113K)个单独的 JSON 文件组成，每个文件包含 100-1000 个观察值。这些观察不是最高级别的，但需要在每个 JSON 中进行一些导航才能访问。
IE。
json["interactions"] 是一个字典列表。

我正在尝试利用 Spark/PySpark(版本 1.1.1)来解析和减少这些数据，但我无法找出将其加载到 RDD 的正确方法，因为它既不是所有记录 > 一个文件(在在这种情况下，我会使用 sc.textFile，尽管这里增加了 JSON 的复杂性)也不是每条记录 > 一个文件(在这种情况下我会使用 sc.wholeTextFiles)。

我最好的选择是使用 sc.wholeTextFiles 然后使用 map (或者在这种情况下是 flatMap？)将多个观察结果从单个文件名键下存储到它们自己的键中？或者有没有更简单的方法来做到这一点，我错过了？

我在这里看到的答案建议只对通过 sc.textFile 加载的所有文件使用 json.loads()，但它似乎对我不起作用，因为 JSON 不是简单的最高级别列表。

Answer 1

以前的答案不会以分布式方式读取文件(请参阅 reference )。为此，您需要并行化 s3 键，然后在如下所示的 flatMap 步骤中读入文件。

import boto3
import json
from pyspark.sql import Row

def distributedJsonRead(s3Key):
    s3obj = boto3.resource('s3').Object(bucket_name='bucketName', key=s3Key)
    contents = json.loads(s3obj.get()['Body'].read().decode('utf-8'))
    for dicts in content['interactions']
        yield Row(**dicts)

pkeys = sc.parallelize(keyList) #keyList is a list of s3 keys
dataRdd = pkeys.flatMap(distributedJsonRead)

Boto3 Reference