使用Spark Scala添加新列使用現有的列

嗨我想添加新列使用DataFrame的每一行中的現有列，我試圖在Spark Scala這樣... df是包含可變數量的列，這些只能在運行時決定。使用Spark Scala添加新列使用現有的列

// 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) 
    })

但這給錯誤是Eclipse本身

無法找到存儲在數據集型編碼器。通過導入spark.implicits._支持原始類型（Int，String等）和Product類型（case類）。將來的發行版中將添加對序列化其他類型的支持。（隱式證據$ 7：org.apache.spark.sql.Encoder [org.apache.spark.sql.Row]）org.apache.spark.sql.Dataset [org.apache。 spark.sql.Row。未指定的值參數證明$ 7。

請幫忙。

來源

2017-10-09 Shiv Dengre

的'import'應'map'之外做來完成。 – Shaido

你能舉出輸入數據和期望輸出的例子嗎？這應該有可能以更有效的方式解決。 – Shaido

它可以更多更好地使用UDF和withColumn阿比

來源

2017-10-09 09:54:35

使用Spark Scala添加新列使用現有的列

回答

相關問題