スパーク - スカラ

Question

test.csv 
name,key1,key2 
A,1,2 
B,1,3 
C,4,3 

とCSVデータ分割私は（データセットまたはRDDとして）このように、このデータを変更したい

whatIwant.csv 
name,key,newkeyname 
A,1,KEYA 
A,2,KEYB 
B,1,KEYA 
B,3,KEYB 
C,4,KEYA 
C,3,KEYB 

私は、readメソッドでデータをロードしました。

val df = spark.read 
      .option("header", true) 
      .option("charset", "euc-kr") 
      .csv(csvFilePath) 

私は（名前、キー1）または（名前、KEY2）のように、各データセットをロードし、そして組合それらを結合によって、しかし、単一の火花セッションでこれをやりたいことができます。 これは何ですか？

---

これらは機能しません。

val df2 = df.select(df("TAG_NO"), df.map { x => (x.getAs[String]("MK_VNDRNM"), x.getAs[String]("WK_ORD_DT")) }) 

val df2 = df.select(df("TAG_NO"), Seq(df("TAG_NO"), df("WK_ORD_DT"))) 

Answer 1

これはexplodeとudfで実現することができます。

scala> val df = Seq(("A", 1, 2), ("B", 1, 3), ("C", 4, 3)).toDF("name", "key1", "key2") 
df: org.apache.spark.sql.DataFrame = [name: string, key1: int ... 1 more field] 

scala> df.show 
+----+----+----+ 
|name|key1|key2| 
+----+----+----+ 
| A| 1| 2| 
| B| 1| 3| 
| C| 4| 3| 
+----+----+----+ 

scala> val explodeUDF = udf((v1: String, v2: String) => Vector((v1, "Key1"), (v2, "Key2"))) 
explodeUDF: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function2>,ArrayType(StructType(StructField(_1,StringType,true), StructField(_2,StringType,true)),true),Some(List(StringType, StringType))) 

scala> df = df.withColumn("TMP", explode(explodeUDF($"key1", $"key2"))).drop("key1", "key2") 
df: org.apache.spark.sql.DataFrame = [name: string, TMP: struct<_1: string, _2: string>] 

scala> df = df.withColumn("key", $"TMP".apply("_1")).withColumn("new key name", $"TMP".apply("_2")) 
df: org.apache.spark.sql.DataFrame = [name: string, TMP: struct<_1: string, _2: string> ... 2 more fields] 

scala> df = df.drop("TMP") 
df: org.apache.spark.sql.DataFrame = [name: string, key: string ... 1 more field] 

scala> df.show 
+----+---+------------+ 
|name|key|new key name| 
+----+---+------------+ 
| A| 1|  Key1| 
| A| 2|  Key2| 
| B| 1|  Key1| 
| B| 3|  Key2| 
| C| 4|  Key1| 
| C| 3|  Key2| 
+----+---+------------+