データフレームに行を再帰的に追加する

Question

私はスパークするのが初めてです。私はとして来るjsonデータを持っています。このデータをハイブテーブルに格納する必要があります。 HttpGet要求ごとに、テーブル内の単一の行になるjsonが返されます。このため、ハイブテーブルのディレクトリにファイルとして単一の行を書き込む必要があります。

しかし、私は小さすぎるファイルを持っていると感じるので、速度と効率が低下します。だから私は再帰的にDataframeに新しい行を追加し、ハイブテーブルのディレクトリに一度に書き込む方法があります。私はこれも私のスパークコードの実行時間を減らすと思う。

例：私は、データフレームは不変であることを理解

for(i <- 1 to 10){ 
newDF = hiveContext.read.json("path") 
df = df.union(newDF) 
} 
df.write() 

。これを達成する方法はありますか？

ご協力いただければ幸いです。ありがとうございました。

Answer 1

あなたは正しい軌道に乗って大部分があり、何がやりたいことはSeq[DataFrame]として、複数の単一レコードを取得し、それらを合併することにより、単一のDataFrameにSeq[DataFrame]を減らすことです。あなたが提供されているコードから行く

：あなたが行くようにHTTPリクエストを実行する場合

val BatchSize = 100 
val HiveTableName = "table" 

(0 until BatchSize). 
map(_ => hiveContext.read.json("path")). 
reduce(_ union _). 
write.insertInto(HiveTableName) 

を別の方法としては、私たちもそれを行うことができます。あなたがHTTPリクエストを行い、データフレームに変換する機能を持っていると仮定しましょう：

def obtainRecord(...): DataFrame = ??? 

あなたはの線に沿って何かを行うことができます。

val HiveTableName = "table" 
val OtherHiveTableName = "other_table" 
val jsonArray = ??? 

val batched: DataFrame = 
    jsonArray. 
    map { parameter => 
     obtainRecord(parameter) 
    }. 
    reduce(_ union _) 
batched.write.insertInto(HiveTableName) 
batched.select($"...").write.insertInto(OtherHiveTableName) 

Answer 2

あなたは明らかにスパークを悪用しています。 Apache Sparkは分析システムであり、データベースAPIではありません。このようにHiveデータベースを修正するためにSparkを使用する利点はありません。分散処理を含むスパーク機能の恩恵を受けることなく、重大なパフォーマンス上の不利益を被るだけです。

代わりに、Hiveクライアントを直接使用してトランザクション操作を実行する必要があります。

Answer 3

データをすべて（たとえばcurlやその他のプログラムを使用してスクリプトを使用して）一括ダウンロードし、最初にファイルに保存することができれば（多くのファイルでは、ディレクトリ全体を一度に読み込むことができます）そのファイル（または複数のファイル）をまとめてロードして処理します。また、一度に1つのレコードではなく、必要なすべてのデータをフェッチするエンドポイントとしてwebapiを確認します。