运行基础PySpark流应用时遭遇MicroBatchExecution错误求助

常见错误原因及对应解决步骤 #

1. 未启动本地Socket服务 #

代码通过socket数据源监听localhost:9999，但无对应Socket服务器发送数据时，会触发流执行错误。

• 解决方法：
  • Windows系统可打开命令提示符，执行nc -lp 9999（需提前安装Netcat）；
  • 或用Python快速启动Socket服务：

    import socket
    
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 9999))
    server_socket.listen(1)
    conn, addr = server_socket.accept()
    while True:
        data = input("输入要发送的内容：")
        conn.send((data + '\n').encode('utf-8'))
    

  • 启动Socket服务后再运行PySpark代码。

2. Checkpoint目录权限或存在性问题 #

• 错误原因：C:/checkpoint/可能不存在，或Windows系统下无C盘根目录写入权限，导致流处理无法创建checkpoint文件；若之前运行失败残留文件，也会引发异常。
• 解决方法：
  • 更换到有权限的用户目录，比如：

    checkpt = 'C:/Users/你的用户名/Documents/checkpoint/'
    

  • 提前创建目录或确保Spark有足够权限自动创建；
  • 若目录已有残留文件，删除后重新运行代码。

3. 数据处理逻辑的潜在问题 #

原代码中split(col('value'),'\\s')生成数组类型字段，直接groupBy('word')在complete输出模式下可能引发类型匹配异常，间接导致流执行错误。

• 优化处理逻辑：

  # 将数组展开为单个单词
  words = lines.select(explode(split(col('value'), '\\s')).alias('word'))
  counts = words.groupBy('word').count()
  

4. Spark环境配置问题 #

• 确保PySpark环境配置正确：
  • 检查SPARK_HOME环境变量是否设置；
  • 确认使用Spark 2.0+版本（仅该版本及以上支持Structured Streaming）；
  • 用spark-submit命令运行代码，避免直接用Python解释器导致依赖缺失：

    spark-submit your_streaming_code.py
    

修正后的完整代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, split, explode

spark = SparkSession.builder.appName('SparkStream').getOrCreate()
# 监听socket流
lines = spark.readStream.format('socket').option('host','localhost').option('port',9999).load()
# 分割并展开单词
words = lines.select(explode(split(col('value'), '\\s')).alias('word'))
# 统计单词数量
counts = words.groupBy('word').count()
# 使用有权限的checkpoint目录
checkpt = 'C:/Users/你的用户名/Documents/checkpoint/'
# 启动流查询
streamingQuery = counts.writeStream.format('console')\
    .outputMode('complete')\
    .trigger(processingTime='1 second')\
    .option('checkpointLocation', checkpt)\
    .start()
streamingQuery.awaitTermination()

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