Spark Rowsの新しいスキーマの定義

Question

私はDataFrameを持っていて、その列の1つにJSONという文字列が含まれています。これまでのところ、JavaRDD.mapメソッドの必要に応じて、Function<Row,Row>()のようにFunctionインターフェイスを実装しました。この関数の中で、私はJSONを解析していて、追加の列がJSONの値から来た新しい行を作成しています。

オリジナルの行：：私の関数を適用した後

+------+-----------------------------------+ 
| id |  json      | 
+------+-----------------------------------+ 
| 1 | {"id":"abcd", "name":"dmux",...} | 
+------------------------------------------+ 

：たとえば返さJavaRDDから新しいデータフレームを作成しようとしたとき

+------+----------+-----------+ 
| id | json_id | json_name | 
+------+----------+-----------+ 
| 1 | abcd  | dmux  | 
+-----------------+-----------+ 

は私がトラブルに実行していますよ。これらの新しい行があるので、私はスキーマを作成する必要があります。スキーマはJSONの構造に大きく依存しているので、スキーマデータを関数から戻して、オブジェクトと一緒に渡す方法を見つけようとしています。 SparkContextが関数に渡されないので、私はbroadcastの変数を使うことができません。

Functionの呼び出し元の行の各列をループする以外に、どのようなオプションがありますか？

Answer 1

StructTypeを作成することができます。これはScalaですが、それは同じように動作します：

val newSchema = StructType(Array(
    StructField("id", LongType, false), 
    StructField("json_id", StringType, false), 
    StructField("json_name", StringType, false) 
)) 

val newDf = sqlContext.createDataFrame(rdd, newSchema) 

ところで、あなたはあなたのrddがタイプRDD[Row]であることを確認する必要があります。