ケースクラスとカラム名エイリアスでリフレクションを使用したSparkデータフレームスキーマ定義

Question

Spark Scalaスクリプトで少し問題が発生しました。基本的には、グループ化やカウントなどの後に集計を行う生データがあります。出力を特定のJSON形式に保存する必要があります。

EDIT：

私は質問を簡素化しようとし、それを書き直し：

私は列名が別名を持っているArray[org.apache.spark.sql.Column]とソースデータフレームからのデータを選択すると、その列名を使用して（または実際にはインデックス）を変数として使用して、行をクラスにマップしようとすると、「タスクは直列化できません」という例外が発生します。

var dm = sqlContext.createDataFrame(Seq((1,"James"),(2,"Anna"))).toDF("id", "name") 

val cl = dm.columns 
val cl2 = cl.map(name => col(name).as(name.capitalize)) 
val dm2 = dm.select(cl2:_*) 
val n = "Name" 
case class Result(Name:String) 
val r = dm2.map(row => Result(row.getAs(n))).toDF 

そして第二部や質問、私は実際にこれらのResultクラスオブジェクトの配列であることを、最終的なスキーマを必要としています。私はまだ分かっていない、どのようにこれを行うにも。

case class Test(var FilteredStatistics: Array[Result]) 
    val t = Test(Array(Result("Anna"), Result("James"))) 

    val t2 = sc.parallelize(Seq(t)).toDF 

    scala> t2.printSchema 
    root 
    |-- FilteredStatistics: array (nullable = true) 
    | |-- element: struct (containsNull = true) 
    | | |-- Name: string (nullable = true) 

TL; DR：期待される結果はそのようなスキーマを持つべきデータフレームの列が別名を持っていると変数が使用されたときにケースのクラスオブジェクトへのデータフレームの列をマッピングする方法

1. 列名？

2. これらのケースクラスオブジェクトを配列に追加するにはどうすればよいですか？

Answer 1

シリアライゼーション問題：ここでの問題はval n = "Name"である：それは、その変数とそれを含む範囲にわたって密接スパークせるRDD変換（dm2.map(...)）に渡された匿名関数、内部で使用され、 cl2も含まれ、タイプはArray[Column]であるため、シリアル化できません。

解決方法は簡単です - インラインn（dm2.map(row => Result(row.getAs("Name")))）を取得するか、シリアライズ可能なコンテキスト（非シリアル化可能なメンバーを含まないオブジェクトまたはクラス）に配置します。