Python自定义Socket与Spark Structured Streaming通信无实时响应问题

• Spark Structured Streaming内置Socket数据源默认以**换行符\n**作为单条消息的分隔标识，你发送的文本未携带换行符时，Spark会持续缓存数据等待分隔符，直到Socket连接断开才会将缓存的所有内容作为单行处理，这就是关闭服务端后才出结果的核心原因。
• 额外场景下TCP默认启用的Nagle算法会攒小数据包批量发送，也可能导致数据推送延迟。

修改Socket服务端代码，新增两个调整：每条消息末尾加换行符作为分隔，可选禁用Nagle算法避免小包延迟。

修复后的Socket服务端代码 #

import socket
server = socket.socket()
# 禁用Nagle算法，关闭小包发送延迟（可选配置，实时性要求高时建议开启）
server.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
host = '0.0.0.0'
port = 9999
server.bind((host, port))
server.listen(2)
client_socket, addr = server.accept()
print("connection established.")

# 发送数据末尾添加换行符，作为Spark的行分隔标识
client_socket.sendall("Text\n".encode())
# 持续发测试数据可以参考以下逻辑：
# import time
# for i in range(20):
#     client_socket.sendall(f"test word {i%5}\n".encode())
#     time.sleep(1)

修改代码后重启Socket服务端和Spark流程序，发送带换行的消息后，Spark会在默认1秒的触发间隔内实时处理数据，控制台会即时输出统计结果，无需等待Socket连接关闭。

内容的提问来源于stack exchange，提问作者이성진