如何流式傳輸到表中：按列 C 和 D 聚合的列 A 和 B 之間的差值。+-------------+-------------------+--+-| Column_A|Column_B |Column_C|Column_D|+-------------+-------------------+--+-|52 |67 |boy |car ||44 |25 |girl |bike ||98 |85 |boy |car ||52 |41 |girl |car |+-------------+-------------------+--+-這是我的嘗試，但它不起作用：difference = streamingDataF.withColumn("Difference", expr("Column_A - Column_B")).drop("Column_A").drop("Column_B").groupBy("Column_C")differenceStream = difference.writeStream\ .queryName("diff_aggr")\ .format("memory").outputMode("append")\ .start()我收到此錯誤：“GroupedData”對象沒有屬性“writeStream”
1 回答
森林海
TA貢獻2011條經驗 獲得超2個贊
根據您希望如何聚合分組數據 - 您可以執行例如
先決條件（如果您尚未設置它們）：
from pyspark.sql import functions as F
from pyspark.sql.functions import *
為：sum
difference = streamingDataF.withColumn("Difference", expr("Column_A - Column_B")).drop("Column_A").drop("Column_B").groupBy("Column_C").agg(F.sum(F.col("Difference")).alias("Difference"))
為：max
difference = streamingDataF.withColumn("Difference", expr("Column_A - Column_B")).drop("Column_A").drop("Column_B").groupBy("Column_C").agg(F.max(F.col("Difference")).alias("Difference"))
然后：
differenceStream = difference.writeStream\
.queryName("diff_aggr")\
.format("memory").outputMode("append")\
.start()
關鍵是 - 如果你這樣做，你也需要通過聚合來減少。如果你想把你的值排序在一起，試試groupBydf.sort(...)
添加回答
舉報
0/150
提交
取消