如何从 spark 中的另一个数据框值重命名数据框列和数据类型？

Question

嗨，我有两个这样的数据框：

import spark.implicits._
import org.apache.spark.sql._

val transformationDF = Seq(
  ("A_IN", "ain","String"),
  ("ADDR_HASH","addressHash","String")
).toDF("db3Column", "hudiColumn","hudiDatatype")

val addressDF=Seq(
("123","uyt"),
("124","qwe")
).toDF("A_IN", "ADDR_HASH")

现在我想重命名列并更改 transformationdf.The hudi 列名称中提到的值的数据类型，transformationDF 中的 hudidatatype 将成为 addressDF 的列名称和数据类型。我试过这样的代码来改变但不起作用：

var db3ColumnName:String =_
var hudiColumnName:String =_
var hudiDatatypeName:String = _

for (row <- transformationDF.rdd.collect)
{   
     db3ColumnName = row.mkString(",").split(",")(0)
      hudiColumnName= row.mkString(",").split(",")(1)
    hudiDatatypeName = row.mkString(",").split(",")(2)
    

addressDF.withColumnRenamed(db3ColumnName,hudiColumnName).withColumn(hudiColumnName,col(hudiColumnName).cast(hudiDatatypeName))
}

现在，当我打印地址 DF 时，更改不会反映出来。

谁能帮我解决这个问题。

Answer 1

当您使用 withColumnRenamed 或 withColumn 时，它 return 是一个新的数据集，因此您应该这样做：

var db3ColumnName: String = null
var hudiColumnName: String = null
var hudiDatatypeName: String = null

for (row <- transformationDF.rdd.collect) {

  db3ColumnName = row.mkString(",").split(",")(0)
  hudiColumnName = row.mkString(",").split(",")(1)
  hudiDatatypeName = row.mkString(",").split(",")(2)

  addressDF = addressDF.withColumnRenamed(db3ColumnName, hudiColumnName).withColumn(hudiColumnName, col(hudiColumnName).cast(hudiDatatypeName))

}

addressDF.printSchema()

打印地址DF 将return:

root
 |-- ain: string (nullable = true)
 |-- addressHash: string (nullable = true)

Answer 2

这是教科书案例，需要使用foldLeft:

val finalDF = transformationDF.collect.foldLeft(addressDF){ case (df, row) =>
  {
    val db3ColumnName = row.getString(0)
    val hudiColumnName = row.getString(1)
    val hudiDatatypeName = row.getString(2)

    df.withColumnRenamed(db3ColumnName, hudiColumnName)
      .withColumn(hudiColumnName, col(hudiColumnName).cast(hudiDatatypeName))
  }
}

Spark 中的数据集是不可变的，每个“修改”数据集的操作实际上 return 都是一个新对象，而调用该操作的对象保持不变。上面的 foldLeft 有效地从 addressDF 开始并将所有转换链接到作为第二个参数列表中的第一个参数传递的中间对象。当前迭代的return值成为下一次迭代的输入。最后一次迭代的 return 值是 foldLeft 本身的 return 值。

如何从 spark 中的另一个数据框值重命名数据框列和数据类型？

How to rename a dataframe column and datatype from another dataframe values in spark?

scala

apache-spark

apache-spark-sql