新しい列としてキーを使用したキーと値のペアのスパークデータフレームを整形する

Question

sparkとscalaの新機能です。キー値のペアであるリストのデータフレームがあるとします。列IDのid変数を新しい列としてマップする方法はありますか？

df.show() 
+--------------------+-------------------- + 
| ids    | vals     | 
+--------------------+-------------------- + 
|[id1,id2,id3]  | null     | 
|[id2,id5,id6]  |[WrappedArray(0,2,4)] | 
|[id2,id4,id7]  |[WrappedArray(6,8,10)]| 

予想される出力は：

+----+----+ 
|id1 | id2| ... 
+----+----+ 
|null| 0 | ... 
|null| 6 | ... 

Answer 1

可能な方法は、新たなデータフレームの列を計算し、行を構築するために、これらの列を使用することです。

import org.apache.spark.sql.functions._ 

val data = List((Seq("id1","id2","id3"),None),(Seq("id2","id4","id5"),Some(Seq(2,4,5))),(Seq("id3","id5","id6"),Some(Seq(3,5,6)))) 

val df = sparkContext.parallelize(data).toDF("ids","values") 

val values = df.flatMap{ 
    case Row(t1:Seq[String], t2:Seq[Int]) => Some((t1 zip t2).toMap) 
    case Row(_, null) => None 
} 

// get the unique names of the columns across the original data 
val ids = df.select(explode($"ids")).distinct.collect.map(_.getString(0)) 

// map the values to the new columns (to Some value or None) 
val transposed = values.map(entry => Row.fromSeq(ids.map(id => entry.get(id)))) 

// programmatically recreate the target schema with the columns we found in the data 
import org.apache.spark.sql.types._ 
val schema = StructType(ids.map(id => StructField(id, IntegerType, nullable=true))) 

// Create the new DataFrame 
val transposedDf = sqlContext.createDataFrame(transposed, schema) 

バッキングデータソースに応じて、カラム名を計算することはかなり安価であることができるが、このプロセスは、データを介して2回通過します。

また、これはDataFramesとRDDの間を行き来します。私は "純粋な" DataFrameプロセスを見ることに興味があります。