如何在每个窗口周期清理Streaming RDD？

在Spark Streaming场景里，窗口操作会持续保留多个批次的RDD，如果不及时清理，很容易导致内存、磁盘资源被大量占用，拖垮任务性能。结合你的代码片段，我整理了几个实用的清理方法，帮你搞定这个问题：

1. 手动调用unpersist()释放窗口RDD缓存 #

Spark的RDD默认会在计算完成后保留缓存（尤其是窗口操作产生的RDD），我们可以在窗口处理逻辑结束后，主动调用unpersist()来强制释放资源。

下面是补全并修改后的代码示例，重点看窗口处理部分：

from pyspark import SparkConf, SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import Row, SQLContext
import sys
import requests
import traceback

conf = SparkConf()
conf.setAppName("TwitterStreamApp")
# 可选：调整检查点清理延迟，加快旧检查点文件的清理
conf.set("spark.streaming.checkpoint.cleanupDelay", "300000")  # 5分钟，单位毫秒
sc = SparkContext(conf=conf)
sc.setLogLevel("ERROR")
ssc = StreamingContext(sc, 5)
# 检查点路径要确保集群所有节点都能访问
ssc.checkpoint("checkpoint_TwitterApp")

# 从socket获取数据流
dataStream = ssc.socketTextStream("localhost", 9001)

def aggregate_tags_count(new_values, total_sum):
    return sum(new_values) + (total_sum or 0)

def process_window_data(rdd):
    if rdd.isEmpty():
        return
    # 这里是你的业务处理逻辑：解析文本、统计标签
    tags_rdd = rdd.flatMap(lambda line: line.split(" ")).filter(lambda word: word.startswith("#"))
    tag_counts_rdd = tags_rdd.map(lambda x: (x, 1)).updateStateByKey(aggregate_tags_count)
    
    # 处理完成后，立即清理该RDD的缓存
    # blocking=True表示同步等待清理完成，确保资源及时释放
    tag_counts_rdd.unpersist(blocking=True)
    
    # 后续的输出/存储逻辑（比如写入数据库、打印结果）
    tag_counts_rdd.foreach(lambda x: print(f"Tag: {x[0]}, Count: {x[1]}"))

# 定义窗口：窗口长度10秒，滑动间隔5秒（和你的批次间隔一致）
window_stream = dataStream.window(10, 5)
window_stream.foreachRDD(process_window_data)

ssc.start()
ssc.awaitTermination()

2. 优化检查点的自动清理策略 #

Spark Streaming的检查点会保存窗口的状态数据，时间久了会积累大量旧文件。通过调整spark.streaming.checkpoint.cleanupDelay参数，可以让系统更快清理过期的检查点数据（默认是1小时，建议根据窗口大小调整，比如窗口10秒的话，设为5分钟就足够）。

3. 避免不必要的持久化 #

如果你的窗口RDD不需要重复使用，就不要手动调用persist()，让Spark自动管理RDD的生命周期。如果确实需要复用RDD，选择合适的存储级别（比如MEMORY_ONLY_SER，序列化存储减少内存占用），用完后记得及时unpersist()。

4. 利用窗口滑动的自动回收机制 #

当窗口滑动后，超出当前窗口范围的旧批次RDD会被Spark标记为可回收对象，等待垃圾回收。但如果你的任务内存压力大，手动调用unpersist()可以比自动回收更快释放资源，避免OOM问题。

内容的提问来源于stack exchange，提问作者Yiming Sun