PySpark WordCount调用reduceByKey遇tuple索引越界错误求助

我参考教程编写了PySpark WordCount程序，代码如下：

import findspark
findspark.init()

# Create SparkSession and sparkcontext
from pyspark.sql import SparkSession
spark = SparkSession.builder\
                    .master("spark://my-ubuntu.xxx.com:7077")\
                    .appName('wordcount')\
                    .getOrCreate()
sc=spark.sparkContext

# Read the input file and Calculating words count
text_file = sc.textFile("peterpan.txt")
counts = text_file.flatMap(lambda line: line.split(" ")) \
                            .map(lambda word: (word, 1)) \
                           .reduceByKey(lambda x, y: x + y)

# Printing each word with its respective count
output = counts.collect()
for (word, count) in output:
    print("%s: %i" % (word, count))
    

# Stopping Spark-Session and Spark context
sc.stop()
spark.stop()

运行时抛出tuple index out of range错误，完整错误栈如下：

PicklingError                             Traceback (most recent call last)
Cell In [9], line 16
     12 # Read the input file and Calculating words count
     13 text_file = sc.textFile("peterpan1.txt")
     14 counts = text_file.flatMap(lambda line: line.split(" ")) \
     15                             .map(lambda word: (word, 1)) \
---> 16                            .reduceByKey(lambda x, y: x + y)
     18 # Printing each word with its respective count
     19 output = counts.collect()

File ~\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\rdd.py:2275, in RDD.reduceByKey(self, func, numPartitions, partitionFunc)
   2252 def reduceByKey(
   2253     self: "RDD[Tuple[K, V]]",
   2254     func: Callable[[V, V], V],
   2255     numPartitions: Optional[int] = None,
   2256     partitionFunc: Callable[[K], int] = portable_hash,
   2257 ) -> "RDD[Tuple[K, V]]":
   2258     """
   2259     Merge the values for each key using an associative and commutative reduce function.
   2260 
   (...)
   2273     [('a', 2), ('b', 1)]
   2274     """
-> 2275     return self.combineByKey(lambda x: x, func, func, numPartitions, partitionFunc)

...

File ~\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\rdd.py:3345, in _prepare_for_python_RDD(sc, command)
   3342 def _prepare_for_python_RDD(sc: "SparkContext", command: Any) -> Tuple[bytes, Any, Any, Any]:
   3343     # the serialized command will be compressed by broadcast
   3344     ser = CloudPickleSerializer()
-> 3345     pickled_command = ser.dumps(command)
   3346     assert sc._jvm is not None
   3347     if len(pickled_command) > sc._jvm.PythonUtils.getBroadcastThreshold(sc._jsc):  # Default 1M
   3348         # The broadcast will have same life cycle as created PythonRDD

File ~\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\serializers.py:468, in CloudPickleSerializer.dumps(self, obj)
    466     msg = "Could not serialize object: %s: %s" % (e.__class__.__name__, emsg)
    467 print_exec(sys.stderr)
--> 468 raise pickle.PicklingError(msg)

PicklingError: Could not serialize object: IndexError: tuple index out of range

读取的peterpan.txt是包含6000多行内容的Project Gutenberg版《彼得·潘》文本文件。

问题原因 #

1. 版本兼容问题：Python 3.11需要Spark 3.3.0及以上版本支持，若Spark版本过低，会触发序列化错误。
2. Lambda序列化异常：远程集群模式下，匿名Lambda函数的跨节点序列化容易出现不稳定情况。
3. 文本分割不严谨：split(" ")会生成空字符串（如连续空格、行首尾空格），可能间接引发序列化过程中的异常。

解决方案 #

1. 版本适配：升级Spark到3.3.0+，或降级Python到3.10及以下版本。
2. 替换Lambda为具名函数：具名函数的序列化逻辑更稳定，避免匿名函数的传输问题。
3. 优化分割逻辑：使用split("\\s+")分割所有空白字符，同时过滤空字符串，提升结果准确性。

修改后的代码 #

import findspark
findspark.init()

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder\
                    .master("spark://my-ubuntu.xxx.com:7077")\
                    .appName('wordcount')\
                    .getOrCreate()
sc = spark.sparkContext

# 定义具名函数替代Lambda
def split_line(line):
    # 分割所有空白字符并过滤空字符串
    return [word for word in line.split("\\s+") if word]

def map_to_pair(word):
    return (word, 1)

def reduce_counts(x, y):
    return x + y

# 读取文件并计算词频
text_file = sc.textFile("peterpan.txt")
counts = text_file.flatMap(split_line) \
                  .map(map_to_pair) \
                  .reduceByKey(reduce_counts)

# 输出结果
output = counts.collect()
for (word, count) in output:
    print(f"{word}: {count}")

# 停止Spark服务
sc.stop()
spark.stop()

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