如何配置Python Feeder以将stdout数据接入Spark DStream？

下面是两种最常用的实现方式，对应不同的场景需求：

方案1：Socket实时传输（适合低延迟实时场景） #

这是Spark Streaming最经典的实时数据源方案，通过Socket将Feeder生成的数据实时推送给Spark。

步骤1：修改Python Feeder脚本，替换stdout输出为Socket发送 #

把原本打印到控制台的逻辑，改成向指定Socket端口发送数据：

# 修改后的Feeder脚本
import socket
import time

# 配置Spark监听的主机和端口
HOST = 'localhost'
PORT = 9999

# 建立Socket连接并持续发送数据
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((HOST, PORT))
    while True:
        # 替换成你的数据生成逻辑
        generated_data = f"sample_data_{int(time.time())}\n"
        # 发送数据（需编码为字节流，加换行符让Spark按行解析）
        s.sendall(generated_data.encode('utf-8'))
        time.sleep(1)  # 控制数据发送频率

步骤2：编写Spark Streaming代码监听Socket端口 #

创建StreamingContext并监听指定端口，接收Feeder发送的数据：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# 初始化Spark上下文（local[2]表示用2个本地线程）
sc = SparkContext("local[2]", "SocketDStreamFeeder")
# 初始化Streaming上下文，批处理间隔设为1秒
ssc = StreamingContext(sc, 1)

# 监听指定Socket端口，获取DStream
data_stream = ssc.socketTextStream("localhost", 9999)

# 这里添加你的DStream处理逻辑，示例为打印每批数据的前10条
data_stream.pprint()

# 启动Streaming任务并等待终止
ssc.start()
ssc.awaitTermination()

运行顺序 #

必须先启动Spark Streaming程序，再启动修改后的Python Feeder脚本，否则Feeder会因无法连接Socket而报错。

方案2：文件目录监听（适合准实时/离线场景） #

如果不需要极低延迟，可以让Feeder将数据写入指定目录，Spark自动监听该目录的新增文件并读取。

步骤1：修改Python Feeder脚本，输出到指定目录的文件 #

避免覆盖已存在的文件，每次生成新的文件供Spark读取：

# 修改后的Feeder脚本
import time
import os

# 配置Spark监听的目录
TARGET_DIR = "/tmp/spark_streaming_input"
os.makedirs(TARGET_DIR, exist_ok=True)

file_index = 0
while True:
    file_index += 1
    # 生成唯一文件名
    file_path = os.path.join(TARGET_DIR, f"feeder_data_{file_index}.txt")
    with open(file_path, 'w', encoding='utf-8') as f:
        # 替换成你的数据生成逻辑
        generated_data = f"sample_data_{int(time.time())}\n"
        f.write(generated_data)
    time.sleep(2)  # 控制文件生成频率

步骤2：编写Spark Streaming代码监听目录 #

通过textFileStream监听指定目录的新增文本文件：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "FileDStreamFeeder")
ssc = StreamingContext(sc, 2)  # 批处理间隔设为2秒，匹配Feeder的文件生成频率

# 监听指定目录，获取DStream
data_stream = ssc.textFileStream("/tmp/spark_streaming_input")

# 示例处理逻辑：打印每批数据
data_stream.pprint()

ssc.start()
ssc.awaitTermination()

关键注意事项 #

• Socket模式下：Spark的socketTextStream是被动监听，Feeder作为主动发送方，必须保证Spark先启动。
• 文件模式下：Spark只会读取新增的文件，不会重复处理已读取过的文件，因此Feeder不要修改已生成的文件。
• 数据格式：Spark DStream默认按行解析数据，所以Feeder输出的每条数据末尾要加换行符\n，避免多条数据被合并成一条。

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