Spark RDD调用toLocalIterator迭代时出现发送迭代器错误求助

碰到这个报错我太熟了！之前在处理Spark RDD遍历的时候也踩过这个坑，咱们一步步来拆解原因和解决办法：

先说说这个错误的根源 #

这个报错本质上和Spark的闭包序列化、版本兼容性脱不了干系：

• Spark在处理闭包（比如你代码里的print(row)操作）时，会用ASM库来分析和清理闭包中的引用，如果ASM版本和Java、Spark版本不匹配，就会抛出这个IllegalArgumentException。
• 另外，toLocalIterator()会把Executor上的数据分批拉到Driver端处理，如果闭包里引用了不可序列化的对象，或者Driver和Executor的环境不一致，也会触发这个错误。

具体解决办法，按优先级来 #

1. 先检查Java和Spark的版本匹配（最常见原因） #

Spark不同版本对Java的支持是有要求的：

• Spark 2.x系列（比如2.4.x）只推荐用Java 8，如果你用了Java 9及以上版本，asm5（Spark 2.x自带的ASM库）没法解析高版本Java的字节码，直接就报错了。
• Spark 3.x系列用的是asm7+，支持Java 8到17，兼容性好很多。

操作步骤：

• 运行java -version查看当前Java版本，要是用的Java 9+且Spark是2.x，要么把Java降级到8，要么把Spark升级到3.x。

2. 换个遍历方式，别再用toLocalIterator()（最省心） #

其实你只是打印数据的话，完全没必要把数据拉到Driver端！直接让Executor自己打印就行：

# 直接在Executor端执行打印，避免数据拉取和序列化问题
df.foreach(lambda row: print(row))

如果确实需要把数据拉到Driver处理，分两种情况：

• 数据量小：直接用collect()拉取全量数据再遍历（别在大数据量场景用，会OOM）：

  rows = df.collect()
  for row in rows:
      print(row)
  

• 数据量大：用mapPartitions()在Executor端分批处理，根本不用拉到Driver：

  def process_batch(partition):
      # 每个partition就是一批数据，在Executor端处理
      for row in partition:
          print(row)
  # 用count()触发Action执行
  df.rdd.mapPartitions(process_batch).count()
  

3. 检查闭包中的序列化问题 #

如果一定要用toLocalIterator()，得确保你的闭包（就是print(row)这段逻辑）里没有引用不可序列化的对象。比如下面这种写法肯定会报错：

# 错误示例：引用了不可序列化的自定义类对象
class MyNonSerializableClass:
    pass

obj = MyNonSerializableClass()
for row in df.rdd.toLocalIterator():
    print(row, obj)  # obj不可序列化，触发报错

解决办法就是把闭包里的不可序列化对象去掉，或者让对象实现Serializable接口：

# 正确示例：只引用可序列化的内容，或者让对象可序列化
for row in df.rdd.toLocalIterator():
    print(row)  # 这里没有外部不可序列化对象，没问题

4. 排查ASM依赖冲突（比较少见） #

如果你的项目里引入了其他第三方库，刚好带了不同版本的ASM，就会和Spark自带的asm5冲突。这时候需要在构建工具（Maven/Gradle）里排除冲突的ASM依赖，比如Maven的写法：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.4.8</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-asm5-shaded</artifactId>
        </exclusion>
    </exclusions>
</dependency>

之后再引入和Spark兼容的ASM版本就行，不过这个情况不多见，优先试前面的方案。

内容的提问来源于stack exchange，提问作者just another profile name