如何流式傳輸?shù)奖碇校喊戳?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()我收到此錯(cuò)誤：“GroupedData”對(duì)象沒(méi)有屬性“writeStream”
1 回答
森林海
TA貢獻(xiàn)2011條經(jīng)驗(yàn) 獲得超2個(gè)贊
根據(jù)您希望如何聚合分組數(shù)據(jù) - 您可以執(zhí)行例如
先決條件（如果您尚未設(shè)置它們）：
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()
關(guān)鍵是 - 如果你這樣做，你也需要通過(guò)聚合來(lái)減少。如果你想把你的值排序在一起，試試groupBydf.sort(...)
添加回答
舉報(bào)
0/150
提交
取消