我有一个Hive表，它位于spark上下文之上。表格格式如下
| key | param1 | Param 2|
-------------------------
| A | A11 | A12 |
| B | B11 | B12 |
| A | A21 | A22 |我想用schema创建一个Dataframe
val dataSchema = new StructType(
Array(
StructField("key", StringType, nullable = true),
StructField("param", ArrayType(
StructType( Array(
StructField("param1", StringType, nullable = true),
StructField("param2", StringType, nullable = true)
)), containsNull = true), nullable = true)
)
)从上表
最后一张table
| key | param |
-------------------------------------------------------------
| A | [{param1:A11, param2:A12},{param1:A11, param2:A12}]|
| B | [{param1:B11, param2:B12}] |我正在使用配置单元上下文（hivecontext.table（“table\u name”）加载表，它返回Dataframe。
scala> val df = hiveContext.table("sample")
df: org.apache.spark.sql.DataFrame = [fk: string, param1: string, param2: string]
scala> val dfStruct = df.select($"key", struct($"param1", $"param2").alias("param"))
dfStruct: org.apache.spark.sql.DataFrame = [fk: string, sub: struct<param1:string,param2:string>]
scala> dfStruct.show
+--+----------+
|fk| param|
+--+----------+
| A| [A11,A12]|
| B| [B11,B12]|
| A| [A21,A22]|
+--+----------+
scala>我正在尝试使用dataframe来转换到上面使用groupby的表。但做不到。
1条答案
按热度按时间i34xakig1#
我发现我自己。
关键是使用
case class而不是structType```case class Param(param1: String, param2:String)
case class Sample(key: String, param:Array[Param])
val df = hiveContext.table("sample_sub")
val SampleDF = df.select($"fk", $"param1", $"param2")
val SampleDFMap = SampleDF.rdd.groupBy(r => r.getAsString)
val SampleJoinRDD = SampleDFMap.map(
r => Sample(r._1.asInstanceOf[String], r._2.map (
row => Param(row(1).asInstanceOf[String],row(2).asInstanceOf[String])
).toArray
)
)
SampleJoinRDD.toDF.toJSON.collect
// Array({"key":"A","param":[{"param1":"A11","param2":"A12"},{"param1":"A21","param2":"A22"}]}, {"key":"B","param":[{"param1":"B11","param2":"B12"}]})