Spark Scalaを使用して既存の列を追加する

Question

こんにちは、このようなSpark Scalaでこれを試しています... dfは可変数の列を含むデータフレームですが、実行時にのみ決定することができます。

// Added new column "docid" 
val df_new = appContext.sparkSession.sqlContext.createDataFrame(df.rdd, df.schema.add("docid", DataTypes.StringType)) 

df_new.map(x => { 
     import appContext.sparkSession.implicits._ 
     val allVals = (0 to x.size).map(x.get(_)).toSeq 
     val values = allVals ++ allVals.mkString("_") 
     Row.fromSeq(values) 
    }) 

しかし、これはエラーを与えているがデータセットに保存されているタイプのエンコーダを見つけることができません自体が

• 日食です。 spark.implicitsをインポートすることで、プリミティブ型（Int、Stringなど）およびProduct型（ケースクラス）がサポートされています。他の型をシリアル化するためのサポートは、将来のリリースで追加されます。
• メソッドマップの引数が不十分です。（暗黙的な証拠$ 7：org.apache.spark.sql.Encoder [org.apache.spark.sql.Row]）org.apache.spark.sql.Dataset [org.apache。 spark.sql.Row]不特定の値パラメータ証拠$ 7。

助けてください。

Answer 1

それは `import`は` map`の外で行われるべきであるUDFとwithColumnアピ