You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

超大数据集下rdd.zipWithIndex()抛出IllegalArgumentException问题

avatar
安全阿岳
2026-7-15
问题描述

在Azure Databricks的Python笔记本中，尝试用rdd.zipWithIndex()给一个2.72GB、约1238951行的文件添加行号时，抛出IllegalArgumentException，任务运行4小时后失败，怀疑是大小限制，需要避免异常的方法和提速方案，考虑过拆分文件。

代码片段 

runKey = "cca2e0f0-bec0-408a-a5cb-341d26e8b7e0" # 每个文件对应唯一ID
filePath = "/mnt/my_file_path/my_file.txt"
rdd = sc.textFile(filePath)
rdd = rdd.zipWithIndex().map(lambda line: "{}{}{}{}{}".format(str(runKey), delimiter, str(line[1]+1), delimiter, line[0]))

错误输出 

File "<command-3893172145851236>", line 26, in OpenFileRDD
    rdd = rdd.zipWithIndex().map(lambda line: "{}{}{}{}{}".format(str(runKey), delimiter, str(line[1]+1), delimiter, line[0]))
  File "/databricks/spark/python/pyspark/rdd.py", line 2524, in zipWithIndex
    nums = self.mapPartitions(lambda it: [sum(1 for i in it)]).collect()
  File "/databricks/spark/python/pyspark/rdd.py", line 967, in collect
    sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
  File "/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1304, in __call__
    return_value = get_return_value(
  File "/databricks/spark/python/pyspark/sql/utils.py", line 117, in deco
    return f(*a, **kw)
  File "/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py", line 326, in get_return_value
    raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 1234573.0 failed 4 times, most recent failure: Lost task 4.3 in stage 1234573.0 (TID 46064376) (10.0.2.5 executor 5455): java.lang.IllegalArgumentException
    at java.nio.CharBuffer.allocate(CharBuffer.java:334)
    at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:810)
    at org.apache.hadoop.io.Text.decode(Text.java:412)
    at org.apache.hadoop.io.Text.decode(Text.java:389)
    at org.apache.hadoop.io.Text.toString(Text.java:280)
    at org.apache.spark.SparkContext.$anonfun$textFile$2(SparkContext.scala:1065)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
    at scala.collection.Iterator.foreach(Iterator.scala:941)
    at scala.collection.Iterator.foreach$(Iterator.scala:941)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
    at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:442)
    at org.apache.spark.api.python.PythonRunner$$anon$2.writeIteratorToStream(PythonRunner.scala:797)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:521)
    at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:2241)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:313)

Driver stacktrace:
    at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2978)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2925)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2919)
    at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
    at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2919)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1357)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1357)
    at scala.Option.foreach(Option.scala:407)
    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1357)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:3186)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:3127)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:3115)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:1123)
    at org.apache.spark.SparkContext.runJobInternal(SparkContext.scala:2500)
    at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1071)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:165)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:125)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:454)
    at org.apache.spark.rdd.RDD.collect(RDD.scala:1069)
    at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:260)
    at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
    at sun.reflect.GeneratedMethodAccessor6189.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
    at py4j.Gateway.invoke(Gateway.java:295)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:251)
    at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.IllegalArgumentException
    at java.nio.CharBuffer.allocate(CharBuffer.java:334)
    at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:810)
    at org.apache.hadoop.io.Text.decode(Text.java:412)
    at org.apache.hadoop.io.Text.decode(Text.java:389)
    at org.apache.hadoop.io.Text.toString(Text.java:280)
    at org.apache.spark.SparkContext.$anonfun$textFile$2(SparkContext.scala:1065)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
    at scala.collection.Iterator.foreach(Iterator.scala:941)
    at scala.collection.Iterator.foreach$(Iterator.scala:941)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
    at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:442)
    at org.apache.spark.api.python.PythonRunner$$anon$2.writeIteratorToStream(PythonRunner.scala:797)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:521)
    at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:2241)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:313)
解决方案

一、避免IllegalArgumentException异常的方法

从错误栈来看，异常根源是Spark读取文件时，某一行长度超过Hadoop Text类型的默认限制，或是字符集解码问题。同时zipWithIndex的内部实现会把所有分区行数收集到Driver，增大了Driver压力。

1. 调整Hadoop配置，支持超长行

在读取文件前设置Hadoop参数，增大Text类型的最大处理长度：

# 增大Text类型的最大字节长度，这里设为2GB
sc._jsc.hadoopConfiguration().set("io.text.record.maxlength", "2147483648")
# 调整文件读取缓冲区大小，提升读取效率
sc._jsc.hadoopConfiguration().set("io.file.buffer.size", "131072")

设置完成后再用sc.textFile(filePath)读取文件。

2. 替换zipWithIndex的分布式实现

自己实现分区级的行号生成，避免把所有分区行数收集到Driver：

# 第一步：统计每个分区的行数
def count_partition_lines(split_idx, iterator):
    return [(split_idx, sum(1 for _ in iterator))]

partition_counts = rdd.mapPartitionsWithIndex(count_partition_lines).collect()

# 第二步：计算每个分区的起始行号偏移量
offset_map = {}
current_offset = 0
for split_idx, line_count in partition_counts:
    offset_map[split_idx] = current_offset
    current_offset += line_count

# 第三步：给每行添加连续行号
def add_line_num(split_idx, iterator):
    start_num = offset_map[split_idx]
    for idx, line in enumerate(iterator):
        yield "{}{}{}{}{}".format(runKey, delimiter, str(start_num + idx + 1), delimiter, line)

rdd = rdd.mapPartitionsWithIndex(add_line_num)

二、提速方案

1. 拆分大文件

将大文件拆分为多个小文件，提升并行处理能力：

# 读取文件后重新分区，分区数建议设为集群核数的2-3倍
df = spark.read.text(filePath).repartition(64)
# 将拆分后的文件写入临时路径
df.write.text("/mnt/tmp/split_files")
# 再对拆分后的文件进行行号添加处理

2. 使用DataFrame替代RDD

DataFrame有Spark Catalyst优化，性能优于Python RDD，可通过两种方式生成行号：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

df = spark.read.text(filePath)

# 方式1：生成全局唯一递增ID（非严格连续，性能高）
df = df.withColumn("line_id", F.monotonically_increasing_id() + 1)

# 方式2：生成严格连续行号（需全局排序，数据量大时慎用）
# 若不需要全局连续，可按分区生成行号优化性能
window = Window.partitionBy().orderBy("value")
df = df.withColumn("line_id", F.row_number().over(window))

# 拼接结果列
df = df.withColumn("result", F.concat(F.lit(runKey), F.lit(delimiter), F.col("line_id"), F.lit(delimiter), F.col("value")))

# 写出结果
df.write.text("/mnt/output/result_files")

3. 调优集群资源与参数

  • 增大Executor资源：比如设置--executor-memory 8G --executor-cores 4，提升单任务处理能力；
  • 调整shuffle分区数：设置spark.sql.shuffle.partitions = 128（值为集群核数的2-3倍），优化shuffle性能；
  • 开启广播小数据：如果runKey这类小数据需要广播，可手动用F.broadcast()标记，减少数据传输。

4. 减少Python-JVM交互开销

Python RDD需要在Python和JVM之间序列化数据，性能损耗大。优先使用Scala笔记本，或全程用DataFrame/Spark SQL处理，避免跨语言交互。

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.07.15 00:03:09