一、spark系统崩溃以后，重启时需要记得:
任务配置必须一致：重启时spark-submit的参数（如 Executor 数量、内存）、代码逻辑、检查点路径必须和崩溃前完全一致，否则会报 “检查点不兼容”；
正确代码：加checkpointLocation
query = parsed_df.writeStream
.foreachBatch(write_to_mysql)
.outputMode("append")
.option("checkpointLocation", "file:///tmp/spark_checkpoint") # 检查点记进度
.start()
第一步：处理 1-10 条数据，写入 MySQL 后，检查点会记录 “Kafka 偏移量到 10，这批数据已处理”；
第二步：崩溃重启后，Spark 读取检查点→知道 “已经处理到 10，该处理 11-20 条”→不会重复写入 1-10 条。
数据库写入支持幂等性
def write_to_mysql(batch_df, batch_id):
# 用INSERT ... ON DUPLICATE KEY UPDATE：重复则覆盖，不新增
batch_df.write.format("jdbc")
.option("url", "jdbc:mysql://localhost:3306/test")
.option("dbtable", "user_orders")
.option("user", "root")
.option("password", "123456")
.option("sql", """
INSERT INTO user_orders (order_id, user_id, amount, order_time)
VALUES (?, ?, ?, ?)
ON DUPLICATE KEY UPDATE -- 幂等核心：重复则更新，不新增
user_id=VALUES(user_id),
amount=VALUES(amount),
order_time=VALUES(order_time)
""")
.mode("append")
.save()
输出模式：
append,upate,complete
append，一个窗口期内，每个输出新增不可变的数据，比如日志收集，不可用于聚合计算类数据输出，因为如果聚合类
update，一个窗口期内，每次输出变化的数据，比如聚合类订单金额总数，一个窗口10秒钟，10秒内金额总数不断增长
complete，一个窗口期内，全量输出数据，这种适合排名前十位数据显示，
水印：
核心作用：清理过期窗口状态，防止 OOM，是窗口聚合的 “标配”；
配置口诀：先加水印，后聚合；水印列 = 窗口列；时长 = 窗口 + 乱序缓冲；
平衡原则：水印时长不是越大越好，也不是越小越好 —— 刚好覆盖 “最大乱序时间” 即可（生产环境通常 5-15 分钟）。
水印的核心要求是：必须加在所有 “有状态操作” 之前，示例如下:
order_watermark_df = parsed_order_df.withWatermark("create_time", "15 minutes") # 水印核心行
4. Spark SQL：清洗+去重+聚合（仅核心逻辑）
order_watermark_df.createOrReplaceTempView("tmp_order")
清洗（过滤取消单）+ 去重（基于order_id）+ 关联商品品类（Hive极简维度表）
clean_sql = """
SELECT DISTINCT o.order_id, o.province, o.pay_amount, o.create_time, g.category_name
FROM tmp_order o
LEFT JOIN (SELECT goods_id, category_name FROM dwd.dim_goods_core WHERE is_valid=1) g
ON o.goods_id = g.goods_id
WHERE o.order_status != 'CANCEL' AND o.pay_amount > 0
"""
clean_order_df = spark.sql(clean_sql)
注意：parsed_order_df先加水印，再运行spark sql
迟到数据
迟到数据是指超过水印时间的数据，采用离线兜底策略，Spark 默认会直接丢弃迟到数据，但可以通过侧输出流把这些数据单独捕获，写入专门的 “迟到数据表”，给数据流标记 “迟到数据标签”，Spark 会把超过水印的订单分流到侧输出流，主流程仍按正常逻辑处理，侧输出流单独存储迟到数据，迟到的数据在第二天凌晨再计算一次，计算完以后将数据更新到