使用PySpark处理实时随机数据无输出，求问题排查

你的代码存在几个核心问题，导致没有预期的数据输出：

错误点 #

• queueStream使用逻辑错误：你仅初始化了一个包含空RDD的列表，后续没有往队列中动态添加实际数据；且用map将空RDD转换为生成器的操作完全不符合Spark Streaming的数据流模型，生成器不会被当作有效数据处理。
• 数据生成方式错误：generate_data里的while True会直接阻塞线程，且这种在算子内部生成数据的方式不适用于Spark Streaming，正确的做法是在Driver端主动往流的数据源队列推送数据。
• 数据处理函数类型不匹配：process_data期望处理RDD，但实际传入的是生成器，调用filter和reduce会直接失败，且这部分代码根本没有执行机会。

修正后的代码 #

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
import time
import random
from queue import Queue
import threading

# 初始化SparkContext和StreamingContext
sc = SparkContext("local[2]", "RandomDataStream")
ssc = StreamingContext(sc, 1)  # 批处理间隔1秒

# 创建队列用于存放RDD，queueStream会监听这个队列
rdd_queue = Queue()

def generate_data():
    while True:
        # 生成一批数据（示例：每次生成5个1-100的随机数）
        data_batch = [random.randint(1, 100) for _ in range(5)]
        print("Generated batch data:", data_batch)
        # 将数据转为RDD后放入队列
        rdd_queue.put(sc.parallelize(data_batch))
        time.sleep(1)  # 和批处理间隔匹配，每秒推送一批数据

def process_data(rdd):
    if not rdd.isEmpty():
        filtered_data = rdd.filter(lambda x: x % 2 == 0)
        sum_of_data = filtered_data.reduce(lambda x, y: x + y)
        print("Sum of even numbers in this batch:", sum_of_data)
        return sum_of_data
    return 0

# 使用queueStream监听队列，作为流数据源
stream = ssc.queueStream(rdd_queue)
processed_stream = stream.transform(process_data)

# 打印处理结果用于调试
processed_stream.foreachRDD(lambda rdd: rdd.foreach(print))

# 启动单独线程生成数据，避免阻塞Spark主线程
data_thread = threading.Thread(target=generate_data)
data_thread.daemon = True  # 设置为守护线程，随主程序退出
data_thread.start()

# 启动流处理上下文
print("Starting Spark Streaming context...")
ssc.start()

# 等待流处理终止
print("Awaiting termination...")
ssc.awaitTermination()

修正说明 #

1. 改用Queue()创建队列，通过独立线程向队列推送RDD，这是queueStream的标准用法——动态向队列添加RDD作为流的数据源。
2. 将数据生成逻辑移到Driver端的线程中，每次生成一批数据并转为RDD放入队列，避免阻塞Spark的执行线程。
3. 调整process_data为接收RDD的函数，用transform算子处理每个批次的RDD，同时增加空RDD判断避免报错。

内容的提问来源于stack exchange，提问作者Cole J Bromfield