PySpark中show()正常但saveAsTable()抛出索引错误问题

我之前碰到过一模一样的情况——DataFrame显示正常、Schema也完全没问题，但执行saveAsTable()时就突然报索引越界。其实核心原因几乎都是数据里藏着格式不匹配的问题，show()只展示了前几行正常数据，后面有行的字段数和Schema定义的不一致，导致序列化时触发了索引错误。下面是一步步的排查和解决方法：

1. 先排查数据行的字段数是否匹配Schema #

你的Schema定义了3个字段（User-ID、Location、Age），但大概率有些行因为分隔符问题（比如Location字段里包含了你用来拆分数据的逗号），导致实际拆分出来的字段数多于或少于3个。用下面的代码快速检查：

# 过滤出字段数不符合Schema的行
invalid_rows = bx_users_df.rdd.filter(lambda row: len(row) != 3)
print(f"格式错误的行数：{invalid_rows.count()}")

# 如果有错误行，查看具体内容
if invalid_rows.count() > 0:
    invalid_rows.take(5).foreach(print)

如果确实存在这类行，你需要调整数据读取逻辑：比如读取CSV时加上option("quote", "\"")来处理包含分隔符的字段，或者用mode("DROPMALFORMED")直接丢弃格式错误的行。

2. 处理列名中的特殊字符 #

你的列名User-ID包含横杠，虽然Spark的Schema能识别，但某些存储后端（比如Parquet和Hive交互时）可能对特殊字符有隐性限制。试试重命名列后再保存：

# 重命名带特殊字符的列
cleaned_df = bx_users_df.withColumnRenamed("User-ID", "UserID")
# 重新尝试保存
cleaned_df.write.format('parquet').mode('overwrite').saveAsTable('bx_user')

3. 清洗异常字符或空值 #

数据里的换行符、不可见字符或者异常空值也可能导致序列化出错，先做简单清洗试试：

from pyspark.sql.functions import regexp_replace, trim

# 去除Location字段中的换行符、制表符和多余空格
cleaned_df = bx_users_df.withColumn(
    "Location", 
    trim(regexp_replace("Location", "[\n\r\t]", ""))
)
# 尝试保存
cleaned_df.write.format('parquet').mode('overwrite').saveAsTable('bx_user')

4. 换一种方式创建表 #

如果上面的方法都不行，可以先把DataFrame保存为Parquet文件，再通过SQL创建表，绕开saveAsTable()可能存在的问题：

# 先保存为Parquet文件到临时路径
bx_users_df.write.format('parquet').mode('overwrite').save('/tmp/bx_user_parquet')

# 用SQL创建外部表
sqlContext.sql("""
    CREATE TABLE bx_user 
    USING parquet 
    LOCATION '/tmp/bx_user_parquet'
""")

附：你的原始执行记录与报错日志 #

执行记录：

>>> sqlContext.sql('select * from bx_users limit 2').show()
+-------+--------------------+----+
|User-ID|            Location| Age|
+-------+--------------------+----+
|      1| nyc, new york, usa|NULL|
|      2|stockton, califor...|  18|
+-------+--------------------+----+
>>> bx_users_df.show(2)
+-------+--------------------+----+
|User-ID|            Location| Age|
+-------+--------------------+----+
|      1| nyc, new york, usa|NULL|
|      2|stockton, califor...|  18|
+-------+--------------------+----+
only showing top 2 rows
>>> bx_users_df.printSchema()
root
 |-- User-ID: string (nullable = true)
 |-- Location: string (nullable = true)
 |-- Age: string (nullable = true)
>>> bx_users_df.write.format('parquet').mode('overwrite').saveAsTable('bx_user')

报错日志：

18/05/19 00:12:36 ERROR util.Utils: Aborting task
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
    process()
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "<stdin>", line 1, in <lambda>
IndexError: list index out of range

    at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
    at org.apache.spark.api.python.PythonRunner$$anon$1.next(PythonRDD.scala:129)
    at org.apache.spark.api.python.PythonRunner$$anon$1.next(PythonRDD.scala:125)
    at org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:43)
    at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
    at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply$mcV$sp(WriterContainer.scala:261)
    at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:260)
    at org.apache.spark.sql.execution.datasources.DefaultWriterContainer$$anonfun$writeRows$1.apply(WriterContainer.scala:260)
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1279)
    at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:266)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:148)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:148)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:242)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
18/05/19 00:12:37 ERROR datasources.DefaultWriterContainer: Task attempt attempt_201805190012_0222_m_000000_0 aborted.
18/05/19 00:12:37 ERROR executor.Executor: Exception in task 0.0 in stage 222.0 (TID 245)
org.apache.spark.SparkException: Task failed while writing rows
    at org.apache.spark.sql.execution.datasources.DefaultWriterContainer.writeRows(WriterContainer.scala:269)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:148)
    at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:148)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:242)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

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