嗨,我有两个这样的数据框:
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中提到的值的数据类型。transformationDF中的hudicolumn名称和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))
}
现在,当我打印 addressDF 时,更改不会反映出来。
谁能帮我这个 。
这是一个要求使用的教科书案例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 中的数据集是不可变的,每个“修改”数据集的操作实际上都会返回一个新对象,而不会更改调用该操作的对象。上面foldLeft
有效地从所有转换开始addressDF
并将所有转换链接到中间对象,这些对象作为第二个参数列表中的第一个参数传递。当前迭代的返回值成为下一次迭代的输入。最后一次迭代的返回值就是foldLeft
自身的返回值。
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句