scala - 在火花的循环中更新列值
问题描述
简要问题:
对于更直接的查询,我想依次遍历所有行,并根据特定行的某些条件为某些变量(a、b、c)分配一些值,然后我将分配其中 1 的值变量到该特定行的列中。
详细:
我想在 Spark 中更新数据框中的列值。更新将是有条件的,其中我将在行上运行一个循环并根据该行的其他列的值更新一列。
我尝试使用 withColumn 方法但出现错误。请建议任何其他方法。withColumn 方法的解析也会有很大帮助。
表:
var table1 = Seq((11, 25, 2, 0), (42, 20, 10, 0)).toDF("col_1", "col_2", "col_3", "col_4")
table1.show()
架构:
+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
| 11| 25| 2| 0|
| 42| 20| 10| 0|
+-----+-----+-----+-----+
我在这里尝试了两种方法:
- withColumn
- i("col_4") = adj_c
在下面的代码中,在不同位置初始化的变量只需要根据条件以这种方式放置
代码:
for(i <- table1.rdd.collect()) {
if(i.getAs[Int]("col_1") > 0) {
var adj_a = 0
var adj_c = 0
if(i.getAs[Int]("col_1") > (i.getAs[Int]("col_2") + i.getAs[Int]("col_3"))) {
if(i.getAs[Int]("col_1") < i.getAs[Int]("col_2")) {
adj_a = 10
adj_c = 2
}
else {
adj_a = 5
}
}
else {
adj_c = 1
}
adj_c = adj_c + i.getAs[Int]("col_2")
table1.withColumn("col_4", adj_c)
//i("col_4") = adj_c
}
}
第一种情况下的错误:
table1.withColumn("col_4", adj_c)
<console>:80: error: type mismatch;
found : Int
required: org.apache.spark.sql.Column
table1.withColumn("col_4", adj_c)
^
我也尝试在这里使用 col(adj_c) ,但它开始失败
<console>:80: error: type mismatch;
found : Int
required: String
table1.withColumn("col_4", col(adj_c))
^
第二种情况下的错误:
(i("col_4") = adj_c)
<console>:81: error: value update is not a member of org.apache.spark.sql.Row
i("col_4") = adj_c
^
我希望输出表是:
+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
| 11| 25| 2| 1|
| 42| 20| 10| 5|
+-----+-----+-----+-----+
请提出可能的解决方案,并在对问题有任何疑问时回复。
请帮我解决这个问题,因为我遇到了问题。任何形式的建议都会非常有帮助。
解决方案
UDF 可以与任何自定义逻辑一起用于计算列值,例如:
val calculateCol4 = (col_1:Int, col_2:Int, col_3:Int) =>
if (col_1 > 0) {
var adj_a = 0
var adj_c = 0
if (col_1 > col_2 + col_3) {
if (col_1 < col_2) {
adj_a = 10
adj_c = 2
}
else {
adj_a = 5
}
}
else {
adj_c = 1
}
println("adj_c: "+adj_c)
adj_c = adj_c + col_2
// added for return correct result
adj_c
}
// added for return correct result
else 0
val col4UDF = udf(calculateCol4)
table1.withColumn("col_4",col4UDF($"col_1", $"col_2", $"col_3"))
推荐阅读
- json - 有没有更快的方法来解析 Java 中有效整数的字符串?
- reactjs - 在选择控件中选取项目不会在页面上保持选中状态。
- groovy - 使用groovy从soapui中具有命名空间的XML节点获取属性值
- python - 如何正确使用 tf.scatter_update 进行 N 维更新?
- python - PyGitHub:获取存储库的提交总数
- c - 检测重复的程序
- breakpoints - 如何在 lldb 中打印断点 ID
- r - S3 函数重载中的省略号应该放在哪里?
- python - NameError:未定义外部变量
- c# - HttpWebRequest 异常无法解析远程名称:'http'