spark sqlで2つのデータフレームを変換する

Question

spark scalaで2つのデータフレームをテーブルとして登録しています。この2つのテーブルから

表1：

+-----+--------+ 
    |id |values | 
    +-----+----- + 
    | 0 | v1 | 
    | 0 | v2 | 
    | 1 | v3 | 
    | 1 | v1 | 
    +-----+----- + 

表2：

+-----+----+--- +----+ 
    |id |v1 |v2 | v3 
    +-----+-------- +----+ 
    | 0 | a1| b1| - | 
    | 1 | a2| - | c2 | 

    +-----+---------+----+ 

は、私は上記の2つのテーブルを使用して新しいテーブルを生成したいです。

表3：

+-----+--------+--------+ 
    |id |values | field | 
    +-----+--------+--------+ 
    | 0 | v1 | a1  | 
    | 0 | v2 | b1  | 
    | 1 | v3 | c2  | 
    | 1 | v1 | a2  | 
    +-----+--------+--------+ 

ここで、V1は、私がScalaでスパークSQLを使用していますフォーム

v1: struct (nullable = true) 
    | |-- level1: string (nullable = true) 
    | |-- level2: string (nullable = true) 
    | |-- level3: string (nullable = true) 
    | |-- level4: string (nullable = true) 
    | |-- level5: string (nullable = true) 

です。

いくつかのSQLクエリを書くか、データフレームにいくつかのスパーク関数を使用することで、目的を達成できますか？ここで

Answer 1

は、この出力を生成します。つまり、あなたが使用できるサンプルコードです：

コードは次のようになります。コンソール上でテストしている間

val df1=sc.parallelize(Seq((0,"v1"),(0,"v2"),(1,"v3"),(1,"v1"))).toDF("id","values") 
val df2=sc.parallelize(Seq((0,"a1","b1","-"),(1,"a2","-","b2"))).toDF("id","v1","v2","v3") 
val joinedDF=df1.join(df2,"id") 
val resultDF=joinedDF.rdd.map{row=> 
val id=row.getAs[Int]("id") 
val values=row.getAs[String]("values") 
val feilds=row.getAs[String](values) 
(id,values,feilds) 
}.toDF("id","values","feilds") 

：

scala> val df1=sc.parallelize(Seq((0,"v1"),(0,"v2"),(1,"v3"),(1,"v1"))).toDF("id","values") 
df1: org.apache.spark.sql.DataFrame = [id: int, values: string] 

scala> df1.show 
+---+------+ 
| id|values| 
+---+------+ 
| 0| v1| 
| 0| v2| 
| 1| v3| 
| 1| v1| 
+---+------+ 


scala> val df2=sc.parallelize(Seq((0,"a1","b1","-"),(1,"a2","-","b2"))).toDF("id","v1","v2","v3") 
df2: org.apache.spark.sql.DataFrame = [id: int, v1: string ... 2 more fields] 

scala> df2.show 
+---+---+---+---+ 
| id| v1| v2| v3| 
+---+---+---+---+ 
| 0| a1| b1| -| 
| 1| a2| -| b2| 
+---+---+---+---+ 


scala> val joinedDF=df1.join(df2,"id") 
joinedDF: org.apache.spark.sql.DataFrame = [id: int, values: string ... 3 more fields] 

scala> joinedDF.show 
+---+------+---+---+---+               
| id|values| v1| v2| v3| 
+---+------+---+---+---+ 
| 1| v3| a2| -| b2| 
| 1| v1| a2| -| b2| 
| 0| v1| a1| b1| -| 
| 0| v2| a1| b1| -| 
+---+------+---+---+---+ 


scala> val resultDF=joinedDF.rdd.map{row=> 
    | val id=row.getAs[Int]("id") 
    | val values=row.getAs[String]("values") 
    | val feilds=row.getAs[String](values) 
    | (id,values,feilds) 
    | }.toDF("id","values","feilds") 
resultDF: org.apache.spark.sql.DataFrame = [id: int, values: string ... 1 more field] 

scala> 

scala> resultDF.show 
+---+------+------+                
| id|values|feilds| 
+---+------+------+ 
| 1| v3| b2| 
| 1| v1| a2| 
| 0| v1| a1| 
| 0| v2| b1| 
+---+------+------+ 

これがあなたの問題かもしれないことを願っています。ありがとう！