スパークでネストされたジョブを実行中

Question

私はPySparkを使用しています。私はs3にgziped jsonファイルのリストを持っています。これは、アクセスして、変換して、パーケットでs3にエクスポートする必要があります。各jsonファイルには約100k行が含まれていますので、パラレル化するとそれほど意味がありません（しかし、並列化することはできます）が、並列化したファイルは約5kです。私のアプローチは、スクリプトにjsonファイルリストを渡すことです - >リスト上で並列化を実行する - >マップを実行します（これは私がブロックされるところです）。 jsonにアクセスして変換して、変換されたjsonからDFを作成し、それをs3に寄木細工としてダンプします。

Answer 1

jsonを分散して読み込むには、言及したようにキーを並列化する必要があります。 s3からの読み取り中にこれを行うには、boto3を使用する必要があります。以下は、そうする方法のスケルトンスケッチです。あなたのユースケースに合わせてdistributedJsonReadを変更する必要があるでしょう。

import boto3 
import json 
from pyspark.sql import Row 

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

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

Boto3参考：http://boto3.readthedocs.org/en/latest/guide/quickstart.html

編集：1対処する：

後で出力ファイルに入力ファイルの1マッピングを、それがマージされた寄木細工のデータセットを持つことに容易になるだろうという可能性がありますと連携。 JSONの1マッピング：これはあなたがそれを行うために必要な方法がある場合しかし、あなたは私はあなたが1をしたい場合は、これらの操作を並列化することはできませんとは思わないこの

for k in keyList: 
    rawtext = sc.read.json(k) # or whichever method you need to use to read in the data 
    outpath = k[:-4]+'parquet' 
    rawtext.write.parquet(outpath) 

ような何かを試みることができます寄木細工のファイル。 Sparkの読み書き機能は、ドライバによって呼び出されるように設計されており、scおよびsqlContextにアクセスする必要があります。これは、1つの寄木細工のディレクトリを持っている可能性が高い別の理由です。