Spark 4.0在Windows11下写入DataFrame到Redis遇连接重置问题求助

你遇到的是Windows平台Spark Python Worker意外崩溃导致的任务失败，这是Windows与Spark Python交互的常见兼容性问题，结合你的场景，可通过以下方向修复：

1. 强制Spark使用Anaconda的Python环境 #

Windows下Spark默认可能调用系统自带Python，和你Anaconda环境不一致（比如redis库仅安装在Anaconda中），导致Worker进程依赖缺失崩溃。

• 临时生效：在VS Code终端运行代码前执行：

  set PYSPARK_PYTHON=C:\你的Anaconda路径\envs\你的环境名\python.exe
  set PYSPARK_DRIVER_PYTHON=C:\你的Anaconda路径\envs\你的环境名\python.exe
  

• 永久生效：在系统环境变量中添加上述两个变量，指向Anaconda的Python可执行文件路径。

2. 优化Redis连接逻辑 #

Windows下Spark Worker进程处理分区时，频繁创建/关闭Redis连接可能引发资源泄漏，改用连接池优化：

def save_to_redis_partition(rows):
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=False)
    r = redis.Redis(connection_pool=pool)
    try:
        for row in rows:
            key = f"user:{row['id']}"
            value = row['json']
            r.set(key, value)
    finally:
        pool.disconnect()

3. 调整Spark Local模式并发数 #

Windows对多进程/线程的资源调度逻辑与Linux不同，local[*]可能启动过多Worker进程导致资源耗尽，限制并发数：

spark = SparkSession.builder \
    .appName("PySpark Redis Demo") \
    .master('local[2]')  # 根据机器配置改为1-4的固定值
    .getOrCreate()

4. 检查Redis与防火墙配置 #

• 确认Redis服务在Windows上正常运行，redis.conf中bind 127.0.0.1未被注释，测试时可将protected-mode设为no；
• 检查Windows防火墙是否允许6379端口通信，可临时关闭防火墙验证，或添加Redis程序的入站/出站规则。

5. 简化RDD数据转换（可选） #

Windows下Spark Row对象转字典可能存在兼容性问题，直接通过索引取值避免转换：

# 去掉rdd.map(lambda x: x.asDict())，直接传递Row对象
df_json.select("id","json").rdd.foreachPartition(save_to_redis_partition)

# 修改取值逻辑
def save_to_redis_partition(rows):
    # ... 连接逻辑 ...
    for row in rows:
        key = f"user:{row[0]}"  # 直接取第0列（id）
        value = row[1]  # 直接取第1列（json）
        r.set(key, value)
    # ... 关闭连接 ...

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