实现窗口实时数据统计程序
from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.state import StateSpec# 初始化Spark Streaming上下文 sc = SparkContext(appName="SparkStreamingAdvanced") ssc = StreamingContext(sc, batchDuration=5) # 设置微批处理的时间间隔为5秒 ssc.checkpoint("checkpoint") # 设置检查点目录，用于状态恢复# 定义状态更新函数 def update_word_count(new_values, running_count):"""状态管理：更新单词计数"""if running_count is None:running_count = 0return sum(new_values, running_count)# 创建一个DStream，从Socket接收数据 # 假设数据通过Socket发送到localhost的9999端口 lines = ssc.socketTextStream("localhost", 9999)# 对接收到的每行数据进行处理 words = lines.flatMap(lambda line: line.split(" ")) # 将每行数据分割为单词 word_pairs = words.map(lambda word: (word, 1)) # 将单词映射为键值对 (word, 1)# 使用状态管理更新单词计数 word_counts = word_pairs.updateStateByKey(update_word_count)# 使用窗口操作统计过去30秒内的单词计数，窗口滑动间隔为10秒 windowed_word_counts = word_pairs.reduceByKeyAndWindow(lambda x, y: x + y, # 窗口内数据的聚合函数lambda x, y: x - y, # 窗口移除数据的聚合函数windowDuration=30, # 窗口持续时间slideDuration=10 # 窗口滑动时间 )# 打印状态管理和窗口操作的结果 word_counts.pprint() windowed_word_counts.pprint()# 启动StreamingContext ssc.start()# 等待程序运行 ssc.awaitTermination()
