Spark写入Netezza时JDBC间歇性连接超时问题求助

使用Spark将DataFrame写入Netezza数据库时，出现间歇性JDBC连接超时问题：

• 数千次执行中仅触发18次异常
• 超时均发生在查询执行140秒以上的场景
• 已尝试配置spark.network.timeout，但无效果
• 需了解相关处理经验，以及能否通过JDBC连接字符串配置解决该问题

-- TypeError: An error occurred while calling o136.save.
 : org.netezza.error.NzSQLException: Connection timed out (Connection timed out)
    at org.netezza.sql.NzConnection.initSocket(NzConnection.java:2899)
    at org.netezza.sql.NzConnection.open(NzConnection.java:293)
    at org.netezza.datasource.NzDatasource.getConnection(NzDatasource.java:719)
    at org.netezza.datasource.NzDatasource.getConnection(NzDatasource.java:706)
    at org.netezza.Driver.connect(Driver.java:155)
    at org.apache.spark.sql.execution.datasources.jdbc.connection.BasicConnectionProvider.getConnection(BasicConnectionProvider.scala:49)
    at org.apache.spark.sql.execution.datasources.jdbc.connection.ConnectionProviderBase.create(ConnectionProvider.scala:102)
    at org.apache.spark.sql.jdbc.JdbcDialect.$anonfun$createConnectionFactory$1(JdbcDialects.scala:122)
    at org.apache.spark.sql.jdbc.JdbcDialect.$anonfun$createConnectionFactory$1$adapted(JdbcDialects.scala:118)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.getQueryOutputSchema(JDBCRDD.scala:63)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:58)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation$.getSchema(JDBCRelation.scala:242)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:37)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:88)
    at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
    at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:98)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:109)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:169)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:95)
    at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
    at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)
    at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)
    at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:584)
    at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:176)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:584)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:560)
    at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:94)
    at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:81)
    at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:79)
    at org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:116)
    at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:860)
    at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:390)
    at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:363)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:247)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    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)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
    at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
    at java.base/java.lang.Thread.run(Thread.java:829)

一、JDBC连接字符串配置优化 #

针对Netezza JDBC驱动，添加以下参数延长超时时间：

• socketTimeout：设置Socket连接超时（毫秒），建议设为大于140秒的值，例如socketTimeout=300000（5分钟）
• connectTimeout：设置初始连接超时时间，同步延长至对应值，例如connectTimeout=300000
• 示例连接串：

jdbc:netezza://<host>:<port>/<dbname>?user=<user>&password=<pass>&socketTimeout=300000&connectTimeout=300000

二、Spark JDBC相关配置调整 #

除spark.network.timeout外，需调整以下JDBC连接池和执行参数：

• spark.sql.jdbc.maxConnections：增大JDBC连接池最大连接数，避免长时间等待获取连接
• spark.sql.jdbc.connectionTimeout：设置Spark获取JDBC连接的超时时间，例如300s
• 示例Spark配置：

spark.conf.set("spark.sql.jdbc.maxConnections", "20")
spark.conf.set("spark.sql.jdbc.connectionTimeout", "300s")

三、业务逻辑优化 #

1. 拆分大任务：将执行时间超140秒的DataFrame拆分为多个小批次写入，降低单次连接的负载和超时概率
2. 优化写入逻辑：使用mode("append")时提前过滤重复数据；覆盖写入时，先删除目标表数据再写入，避免锁表导致超时
3. 增加重试机制：针对NzSQLException添加重试逻辑，示例代码：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from org.netezza.error import NzSQLException

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type(NzSQLException))
def write_to_netezza(df, url, properties):
    df.write.jdbc(url=url, table="<table_name>", mode="append", properties=properties)

四、Netezza服务器端配置检查 #

联系DBA确认以下设置：

• 数据库max_connections参数是否充足，避免连接耗尽
• 服务器端连接超时设置，是否存在短于140秒的限制
• 网络防火墙是否有针对长连接的拦截规则

内容的提问来源于stack exchange，提问作者PAULO CRISTIANO KLEIN