基于Spark结构化流中的多个条件从另一列更新列值 [英] Update column value from another columns based on multiple conditions in spark structured streaming
本文介绍了基于Spark结构化流中的多个条件从另一列更新列值的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我希望基于多个条件使用另外两列更新一列中的值。对于Eg-流如下:
+---+---+----+---+
| A | B | C | D |
+---+---+----+---+
| a | T | 10 | 0 |
| a | T | 100| 0 |
| a | L | 0 | 0 |
| a | L | 1 | 0 |
+---+---+----+---+
我得到的是多个条件,如-
(B="T"&;&;C>20)OR(B="L"&;&;C=0)
"T"
、20
、"L"
和0
值是动态的。AND/OR
运算符也在运行时提供。我希望只要条件成立,D = 1
就应该保持D = 0
。条件的数量也是动态的。
我尝试将其与spark-sql
中的UPDATE
命令一起使用,即UPDATE df SET D = '1' WHERE CONDITIONS
。但该公司表示,目前还不支持该更新。生成的数据帧应为-
+---+---+----+---+
| A | B | C | D |
+---+---+----+---+
| a | T | 10 | 0 |
| a | T | 100| 1 |
| a | L | 0 | 1 |
| a | L | 1 | 0 |
+---+---+----+---+
我有什么方法可以做到这一点吗?
推荐答案
我希望您使用的是Python.也会为Scala发布同样的帖子!使用udf
PYTHON
>>> df.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 0|
| a| L| 0| 0|
| a| L| 1| 0|
+---+---+---+---+
>>> def get_column(B, C):
... return int((B == "T" and C > 20) or (B == "L" and C == 0))
...
>>> fun = udf(get_column)
>>> res = df.withColumn("D", fun(df['B'], df['C']))>>> res.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 1|
| a| L| 0| 1|
| a| L| 1| 0|
+---+---+---+---+
SCALA
scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._
scala> df.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 0|
| a| L| 0| 0|
| a| L| 1| 0|
+---+---+---+---+
scala> def get_column(B : String, C : Int) : Int = {
| if((B == "T" && C > 20) || (B == "L" && C == 0))
| 1
| else
| 0
| }
get_column: (B: String, C: Int)Int
scala> val fun = udf(get_column _)
fun: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function2>,IntegerType,Some(List(StringType, IntegerType)
))
scala> val res = df.withColumn("D", fun(df("B"), df("C")))
res: org.apache.spark.sql.DataFrame = [A: string, B: string ... 2 more fields]
scala> res.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 1|
| a| L| 0| 1|
| a| L| 1| 0|
+---+---+---+---+
您还可以像这样使用case
when
和otherwise
:
PYTHON
>>> df.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 0|
| a| L| 0| 0|
| a| L| 1| 0|
+---+---+---+---+
>>> new_column = when(
(col("B") == "T") & (col("C") > 20), 1
).when((col("B") == "L") & (col("C") == 0), 1).otherwise(0)
>>> res = df.withColumn("D", new_column)
>>> res.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 1|
| a| L| 0| 1|
| a| L| 1| 0|
+---+---+---+---+
SCALA
scala> df.show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 0|
| a| L| 0| 0|
| a| L| 1| 0|
+---+---+---+---+
scala> val new_column = when(
| col("B") === "T" && col("C") > 20, 1
| ).when(col("B") === "L" && col("C") === 0, 1 ).otherwise(0)
new_column: org.apache.spark.sql.Column = CASE WHEN ((B = T) AND (C > 20)) THEN 1 WHEN ((B = L) AND (C = 0)) THEN 1 ELSE 0 END
scala> df.withColumn("D", new_column).show()
+---+---+---+---+
| A| B| C| D|
+---+---+---+---+
| a| T| 10| 0|
| a| T|100| 1|
| a| L| 0| 1|
| a| L| 1| 0|
+---+---+---+---+
这篇关于基于Spark结构化流中的多个条件从另一列更新列值的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文