PySpark 3.4.1结构化流：最新记录去重与累计聚合实现方案咨询

针对你的需求，下面给出适配PySpark 3.4.1的完整实现步骤，包括去重、累计聚合以及Sink和输出模式的选择：

一、需求拆解与核心思路 #

1. 去重逻辑：对每个id，仅保留updated_at最新的记录，同时支持后续新数据的更新
2. 累计聚合：基于去重后的数据集，按user_id累加total_time_saved，维护每个用户的累计值

二、分步代码实现 #

1. 流数据去重（保留最新记录） #

利用PySpark 3.4+新增的dropDuplicates参数keep="last"，结合水印（Watermark）管理状态，实现高效去重：

# 假设原始流DataFrame为stream_df
# 添加水印：设置允许的最大数据延迟时间（根据业务调整，示例为1小时）
stream_with_watermark = stream_df.withWatermark("updated_at", "1 hour")

# 按id去重，保留每个id对应的最新（updated_at最晚）记录
deduplicated_stream = stream_with_watermark.dropDuplicates(["id"], keep="last")

• 水印作用：自动清理超过延迟阈值的旧状态，避免状态存储无限膨胀
• keep="last"：直接保留每个id分组中最后（最新）的记录，无需额外窗口函数，简化逻辑

2. 按user_id累计聚合 #

基于去重后的流，执行有状态的累计聚合：

from pyspark.sql.functions import sum

# 按user_id分组，累计求和total_time_saved
aggregated_stream = deduplicated_stream.groupBy("user_id") \
    .agg(sum("total_time_saved").alias("cumulative_total_time_saved"))

三、输出模式（OutputMode）选择 #

根据业务场景选择对应的输出模式：

• update模式：仅输出累计值发生变化的user_id结果，性能最优，适合增量更新场景
• complete模式：每次触发都输出所有user_id的完整累计值，适合需要全量数据的场景

四、Sink实现（针对MongoDB/Kafka） #

1. 输出到MongoDB（用foreachBatch实现Upsert） #

通过foreachBatch在批处理层面实现MongoDB的更新/插入逻辑，确保累计值实时更新：

import pymongo

def write_to_mongo(batch_df, batch_id):
    # 建立MongoDB连接
    client = pymongo.MongoClient("mongodb://your_host:27017/")
    db = client["your_database"]
    coll = db["user_time_aggregates"]
    
    # 将批处理数据转为字典列表
    records = batch_df.toPandas().to_dict("records")
    
    # 批量执行Upsert：存在则更新累计值，不存在则插入
    for rec in records:
        coll.update_one(
            {"user_id": rec["user_id"]},
            {"$set": {"cumulative_total_time_saved": rec["cumulative_total_time_saved"]}},
            upsert=True
        )
    client.close()

# 启动流查询
query = aggregated_stream.writeStream \
    .foreachBatch(write_to_mongo) \
    .outputMode("update") \
    .trigger(processingTime="1 minute")  # 触发间隔根据业务调整
    .start()

query.awaitTermination()

2. 输出到Kafka #

直接用Kafka Sink输出聚合结果，将数据序列化为JSON格式：

aggregated_stream.selectExpr(
    "cast(user_id as string) as key",
    "to_json(struct(*)) as value"
).writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your_broker:9092") \
    .option("topic", "user_time_agg_topic") \
    .outputMode("update") \
    .trigger(processingTime="1 minute") \
    .start()

五、完整示例（含数据源读取） #

以MongoDB为数据源的完整代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum
import pymongo

def write_to_mongo(batch_df, batch_id):
    client = pymongo.MongoClient("mongodb://your_host:27017/")
    db = client["your_database"]
    coll = db["user_time_aggregates"]
    records = batch_df.toPandas().to_dict("records")
    for rec in records:
        coll.update_one(
            {"user_id": rec["user_id"]},
            {"$set": {"cumulative_total_time_saved": rec["cumulative_total_time_saved"]}},
            upsert=True
        )
    client.close()

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("UserTimeCumulativeAgg") \
    .config("spark.mongodb.input.uri", "mongodb://your_host:27017/your_database.source_collection") \
    .getOrCreate()

# 读取MongoDB流数据
stream_df = spark.readStream \
    .format("mongodb") \
    .option("collection", "source_collection") \
    .option("spark.mongodb.input.readPreference.name", "primaryPreferred") \
    .load()

# 去重
stream_with_watermark = stream_df.withWatermark("updated_at", "1 hour")
deduplicated_stream = stream_with_watermark.dropDuplicates(["id"], keep="last")

# 累计聚合
aggregated_stream = deduplicated_stream.groupBy("user_id") \
    .agg(sum("total_time_saved").alias("cumulative_total_time_saved"))

# 输出到MongoDB
query = aggregated_stream.writeStream \
    .foreachBatch(write_to_mongo) \
    .outputMode("update") \
    .trigger(processingTime="1 minute") \
    .start()

query.awaitTermination()

内容的提问来源于stack exchange，提问作者Subhadip Sahoo