Spark任务间歇性失败并触发Connection Reset Error问题排查求助

从你的描述和线程栈来看，问题的核心出在SQL Server JDBC驱动读取大字段（Large Object）时的阻塞，结合你的环境和已尝试的方案，我整理了以下针对性的解决思路：

1. 优先排查大字段类型并升级JDBC驱动 #

从线程栈里的PLPInputStream.readBytesInternal可以明确，任务卡住是因为正在读取SQL Server的大字段（比如VARCHAR(MAX)、NVARCHAR(MAX)、VARBINARY(MAX)这类PLP类型字段）。旧版本的mssql-jdbc驱动在处理这类字段时存在性能或死锁问题：

• 验证大字段影响：临时修改select_query，将大字段替换为截断版本（比如LEFT(large_column, 100)）或者直接排除，测试任务是否还会卡住。如果问题消失，说明大字段是元凶。
• 升级JDBC驱动：你当前使用的mssql-jdbc:9.2.1.jre8是2021年的版本，存在不少PLP相关的bug。建议升级到最新的12.x版本（比如mssql-jdbc:12.4.2.jre8），微软在后续版本中修复了大量大字段读取的稳定性问题。

2. 优化Spark JDBC连接与分区策略 #

调整JDBC连接参数 #

在Spark DataFrame Reader中添加以下参数，优化大结果集的读取逻辑：

• selectMethod=cursor：默认direct模式会一次性将结果集加载到内存，大表容易导致超时；cursor模式采用游标逐批读取，降低内存压力和连接阻塞概率。
• fetchSize=1000：控制每次从数据库拉取的行数，避免一次性拉取过多数据导致连接卡住。
• responseBuffering=adaptive：让驱动根据数据量自动调整缓冲策略，减少Socket读取阻塞。
• 在JDBC URL中添加超时参数：url="jdbc:sqlserver://xxx;loginTimeout=30;queryTimeout=600;socketTimeout=600000;"（注意是在URL里设置，不是Spark的option），强制Socket超时避免无限等待。

示例调整后的Reader代码：

val table_source = spark.read 
.format("com.microsoft.sqlserver.jdbc.spark") 
.option("Driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") 
.option("dbtable", select_query) 
.option("inferschema", "true") 
.option("url", "jdbc:sqlserver://xxx;loginTimeout=30;queryTimeout=600;socketTimeout=600000;") 
.option("user", ds_user) 
.option("password", ds_pass) 
.option("numPartitions", num_partitions) 
.option("partitionColumn", "RNO") 
.option("lowerBound", 0) 
.option("upperBound", rowCount)
.option("selectMethod", "cursor")
.option("fetchSize", "1000")
.option("responseBuffering", "adaptive")
.load()

优化分区策略 #

• 检查RNO字段的数据分布：如果RNO的取值不均匀（比如大部分数据集中在某个区间），会导致部分分区数据量远大于其他分区，卡住的任务就是处理这些大分区。可以换一个分布更均匀的字段作为partitionColumn，或者调整lowerBound和upperBound匹配实际数据的最小值和最大值，避免分区倾斜。
• 如果分区数设置不合理，尝试调整numPartitions（比如从8调整到16或4），平衡每个任务的数据量。

3. 优化Delta Lake写入配置 #

写入阶段的性能瓶颈也可能间接导致读取阶段的阻塞（比如数据处理完无法及时写入，导致Reader线程等待）：

• 设置maxRecordsPerFile=10000：控制每个Delta文件的大小，避免生成超大文件，提升写入并行度。
• 开启Spark的动态分区写入：spark.sql.sources.partitionOverwriteMode=dynamic（如果你的数据是按分区写入的），减少不必要的文件操作。
• 检查EMR到S3的网络性能：确保集群开启了S3 Transfer Acceleration，或者VPC配置没有限制S3的访问带宽，避免写入S3时的网络瓶颈。

4. 调整Spark与JVM的参数 #

• 启用Spark的JDBC重试机制：设置spark.sql.jdbc.retryOnTimeout=true（Spark 3.1+支持），让Spark在遇到连接超时或重置错误时自动重试任务，减少因偶发网络问题导致的失败。
• 调整JVM Socket超时参数：在EMR集群的Spark配置中添加--conf spark.executor.extraJavaOptions="-Dsun.net.client.defaultConnectTimeout=30000 -Dsun.net.client.defaultReadTimeout=600000"，强制Socket读取超时，避免线程无限期卡住。
• 避免连接共享：设置spark.sql.jdbc.numConnectionsPerPartition=1，每个分区使用独立的数据库连接，减少JDBC驱动的线程锁竞争（从你的线程栈看，存在TDSReader的监视器锁，连接共享可能加剧这个问题）。

5. 替代方案：先导出到中间存储 #

如果以上方案都无法解决，考虑用SSIS先将SQL Server数据导出到S3的Parquet文件，再用Spark读取Parquet并转换为Delta格式。因为你提到SSIS只需要20分钟就能完成数据拉取，说明这种方式更高效，避免Spark JDBC的瓶颈。

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