使用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)
    }) 

但这个错误是eclipse本身

• 无法找到存储在数据集中的类型的编码器。通过导入spark.implicits.支持原始类型（Int，String等）和产品类型（case类）_将在未来版本中添加对其他类型序列化的支持。
• 方法映射没有足够的参数：（隐式证据7美元：org.apache.spark.sql.编码器[org.apache.spark.sql.行]）org.apache.spark.sql.数据集[org.apache.spark.sql.行]。未指定的值参数证据7美元。

请帮忙。

Answer 1

函数对象中的concat_ws会有所帮助。

此代码添加了docid字段

df = df.withColumn("docid", concat_ws("_", df.columns.map(df.col(_)):_*))

假设df的所有列都是字符串。