PySpark: как читать много файлов JSON, несколько записей на файл

Question

У меня есть большой набор данных, хранящийся в ведре S3, но вместо того, чтобы быть одним большим файлом, он состоит из множества (113K, если быть точным) отдельных файлов JSON, каждая из которых содержит 100-1000 наблюдений. Эти наблюдения не находятся на самом высоком уровне, но требуют некоторой навигации в каждом JSON для доступа. i.e. json ["взаимодействия"] - это список словарей.

Я пытаюсь использовать Spark/PySpark (версия 1.1.1) для анализа и сокращения этих данных, но я не могу найти правильный способ загрузить его в RDD, потому что это не все записи> один файл (в этом случае я бы использовал sc.textFile, хотя добавили здесь сложность JSON), ни каждая запись> один файл (в этом случае я бы использовал sc.wholeTextFiles).

Является ли мой лучший вариант использовать sc.wholeTextFiles, а затем использовать карту (или в данном случае flatMap?), Чтобы вытащить несколько наблюдений из-под одного ключа имени на свой собственный ключ? Или есть более простой способ сделать это, чтобы я отсутствовал?

Я видел ответы здесь, в которых предлагается только использовать json.loads() для всех файлов, загружаемых через sc.textFile, но похоже, что это не сработает для меня, потому что JSON не являются простыми на самом высоком уровне списки.

Answer 1

Имя вводит в заблуждение (потому что оно сингулярно), но sparkContext.textFile() (по крайней мере, в случае с Scala) также принимает имя каталога или шаблон подстановки, поэтому вы можете просто сказать textFile("/my/dir/*.json").

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) 

Я новичок, чтобы зажечь себя, так что я интересно, если есть лучший способ использовать dataframes с большим количеством файлов s3, но до сих пор это работает для меня.

Answer 3

Предыдущие ответы не будут читать файлы распределенным способом. Для этого вам нужно будет распараллелить ключи 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 Ссылка: http://boto3.readthedocs.org/en/latest/guide/quickstart.html