PySpark批量插入Teradata遇BatchUpdateException问题排查与解决

使用PySpark将HBase数据导入Teradata时，当Spark DataFrame限制为3000-5000行（如df = df.limit(3000)），插入操作可成功执行；但取消行限制并设置Teradata的batchsize为1000时，触发如下错误：

java.sql.BatchUpdateException: [Teradata JDBC Driver] [TeraJDBC 17.10.00.27] 
[Error 1338] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. 
The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.

完整代码片段 #

df = df.toDF(*renamed_columns)
df = df.withColumnRenamed("data_ROW", "ROW")
df.show(1, False)
df = df.limit(5)

teraDataIp = configFile["teraDataDevIP"]
teraDataBaseName = configFile["teraDataDbName"]
jdbc_url = "jdbc:teradata://{}/DATABASE={},tmode=ANSI,charSet=UTF8,SSLMODE=DISABLE".format(teraDataIp, teraDataBaseName)

if "1" in hbaseTableStr:
    teradata_table = "____"
elif "2" in hbaseTableStr:
    teradata_table = "____"
elif "3" in hbaseTableStr:
    teradata_table = "____"
elif "4" in hbaseTableStr:
    teradata_table = "____"

print("Teradata Table Name is: ", teradata_table)
df = df.repartition(10)

df.write.format("jdbc") \
    .mode("append") \
    .option("driver", "com.teradata.jdbc.TeraDriver") \
    .option("url", jdbc_url) \
    .option("user", ____) \
    .option("password", ____) \
    .option("dbtable", teradata_table) \
    .option("batchsize", 1000).save()

执行命令 #

spark-submit --conf "spark.driver.extraClassPath=[REDACTED]/hbase/lib/*" \
--jars [REDACTED]/hbase_connectors/hbase-spark-protocol-shaded.jar,\
[REDACTED]/lib/terajdbc4.jar,\
[REDACTED]/tdgssconfig.jar \
[REDACTED]/sparkHbase_v2.py

完整错误栈 #

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
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:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)

Caused by: com.teradata.jdbc.jdbc_4.util.JDBCException 
[Teradata Database] [TeraJDBC 17.10.00.27] [Error 1338] [SQLState HY000] 
A failure occurred while executing a PreparedStatement batch request. 
The presentation of the failure can be found in the exception chain which is accessible with getNextException().
Details of the failure can be found in the exception chain which is accessible with getNextException().
    at com.teradata.jdbc.jdbc_4.util.ErrorFactory.makeBatchUpdateException(ErrorFactory.java:198)
    at com.teradata.jdbc.jdbc_4.statemachine.StatementReceiveState.batchUpdateRowCount(StatementReceiveState.java:1406)
    at com.teradata.jdbc.jdbc_4.statemachine.StatementReceiveState.batchUpdateRowCount(StatementReceiveState.java:1389)
    at com.teradata.jdbc.jdbc_4.statemachine.StatementReceiveState.batchUpdateRowCount(StatementReceiveState.java:1371)

Caused by:
org.apache.spark.SparkException Job aborted due to stage failure:
Task 0 in stage 1 (TID 1) failed for unknown reasons
org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:630)

Caused by:
java.sql.BatchUpdateException Batch entry 0 insert into table_name values (?, ?, ?) was aborted.
Call getNextException() to see other errors in the batch.
    at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.run(DataSourceRDD.scala$anon$1.scala$166)

已尝试操作 #

• 插入小数据集可正常执行
• 检查Teradata数据库权限无问题

核心疑问：为何小批量限制插入正常，但设置batchsize插入大数据集时失败？如何正确将大型Spark DataFrame从PySpark插入Teradata而不触发该BatchUpdateException？

1. 脏数据触发约束违规：小批量数据刚好避开了违反Teradata表约束的记录，而大数据集中存在非法行（如字段长度超限、数据类型不匹配、主键重复、非空字段为空等），批量插入时Teradata会因单条非法记录终止整个批次。
2. JDBC批次处理兼容性问题：Spark的JDBC批量逻辑与Teradata驱动的批次机制冲突，当批次内出现异常时，驱动无法精准返回错误行信息，直接导致整个任务失败。
3. 资源或超时问题：大数据集的单分区或单批次数据量过大，引发Teradata端超时、资源耗尽，进而触发批次执行失败。

1. 定位并清洗脏数据 #

在插入前对DataFrame做全量校验，找出非法记录并处理：

• 校验字段长度：

  from pyspark.sql.functions import length, col
  # 示例：检查"name"字段是否超过表定义的50字符限制
  invalid_length_rows = df.filter(length(col("name")) > 50)
  invalid_length_rows.show()
  

• 检查非空字段：

  null_rows = df.filter(col("required_column").isNull())
  null_rows.show()
  

• 排查主键重复：

  duplicate_keys = df.groupBy("primary_key").count().filter(col("count") > 1)
  duplicate_keys.show()
  

对非法数据进行截断、填充默认值或过滤后，再执行插入操作。

2. 调整批量插入参数 #

• 减小batchsize：将当前的1000调至200-500，降低单批次出错概率，同时便于定位错误行。
• 优化分区数：根据数据总量合理设置numPartitions，避免单分区数据量过大：

  df.write.format("jdbc") \
      .mode("append") \
      .option("driver", "com.teradata.jdbc.TeraDriver") \
      .option("url", jdbc_url) \
      .option("user", "xxx") \
      .option("password", "xxx") \
      .option("dbtable", teradata_table) \
      .option("batchsize", 500) \
      .option("numPartitions", 20) \
      .option("retryOnTimeout", "true") \
      .save()
  

3. 配置Teradata驱动的批量错误处理 #

在JDBC URL中添加batchmode=perStatement参数，让驱动逐条执行批次内语句，精准返回错误行：

jdbc_url = "jdbc:teradata://{}/DATABASE={},tmode=ANSI,charSet=UTF8,SSLMODE=DISABLE,batchmode=perStatement".format(teraDataIp, teraDataBaseName)

该参数会略微降低性能，但能快速定位问题根源。

4. 使用Teradata专用Spark连接器 #

替换通用JDBC，使用Teradata官方提供的Spark连接器（Teradata Connector for Spark），它针对Teradata特性做了优化，支持更稳定的批量插入、自动数据类型适配等功能。

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