PySpark:如何读取许多JSON文件,每个文件多个记录

Question

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

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

我最好选择使用sc.wholeTextFiles,然后使用地图(或者在这种情况下使用flatMap？)将多个观察结果从单个文件名密钥存储到自己的密钥中？或者有一种更容易的方法来做到这一点,我错过了？

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

Answer 1

以前的答案不会以分布式方式读取文件(参见参考资料).为此,您需要并行化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参考

Answer 2

怎样使用DataFrames？

testFrame = sqlContext.read.json('s3n://<bucket>/<key>') 从一个文件中 给您想要的东西？

每个观察都具有相同的“列”（键数）吗？

如果是这样，您可以使用boto列出要添加的每个对象，将它们读入并相互合并。

from pyspark.sql import SQLContext
import boto3
from pyspark.sql.types import *
sqlContext = SQLContext(sc)

s3 = boto3.resource('s3')
bucket = s3.Bucket('<bucket>')

aws_secret_access_key = '<secret>'
aws_access_key_id = '<key>'

#Configure spark with your S3 access keys
sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", aws_access_key_id)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", aws_secret_access_key)
object_list = [k for k in bucket.objects.all() ]
key_list = [k.key for k in bucket.objects.all()]

paths = ['s3n://'+o.bucket_name+'/'+ o.key for o in object_list ]

dataframes = [sqlContext.read.json(path) for path in paths]

df = dataframes[0]
for idx, frame in enumerate(dataframes):
    df = df.unionAll(frame)

我是新来的人，所以我想知道是否有更好的方法来使用带有很多s3文件的数据帧，但是到目前为止，这对我来说是可行的。