如何在Spark中调用SQL Server存储过程？

Spark完全可以调用SQL Server存储过程，你之前的尝试失败大概率是没匹配JDBC调用存储过程的正确姿势，或是没处理好更新类操作的特性。以下是具体实现方案：

一、仅执行更新操作的存储过程（无返回结果） #

这类场景（比如更新最后执行时间表）适合直接通过JDBC连接执行，绕开Spark的分布式查询逻辑：

import java.sql.DriverManager
import org.apache.spark.sql.SparkSession

// 初始化SparkSession
val spark = SparkSession.builder().getOrCreate()
// 配置SQL Server JDBC连接参数
val jdbcUrl = "jdbc:sqlserver://your-server:1433;databaseName=your-db;user=your-username;password=your-password"
val driverClass = "com.microsoft.sqlserver.jdbc.SQLServerDriver"

// 加载驱动并获取连接
Class.forName(driverClass)
val conn = DriverManager.getConnection(jdbcUrl)

// 调用存储过程（无参数）
val callStmt = conn.prepareCall("{call dbo.UpdateLastExecTime()}")
// 如果存储过程带参数，例如{call dbo.UpdateLastExecTime(?, ?)}，需先设置参数
// callStmt.setString(1, "your-param1")
// callStmt.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()))

// 执行更新操作
callStmt.execute()

// 关闭资源
callStmt.close()
conn.close()

二、带结果集返回的存储过程 #

如果存储过程同时包含更新逻辑和结果输出，可通过Spark JDBC的CALL语法调用：

方式1：Spark SQL直接调用 #

spark.sql("""
    CALL dbo.YourProcedureWithResult(?)
""", "your-param-value")

方式2：JDBC读入结果集为DataFrame #

val resultDF = spark.read.format("jdbc")
  .option("url", jdbcUrl)
  .option("dbtable", "{call dbo.YourProcedureWithResult(?)}")
  .option("user", "your-username")
  .option("password", "your-password")
  .option("driver", driverClass)
  .option("queryTimeout", "30") // 设置超时时间
  .load()

// 处理结果集
resultDF.show()

注意：如果存储过程返回多个结果集，Spark默认仅读取第一个，需额外通过JDBC连接手动处理后续结果集。

三、为什么之前的方法失败？ #

• 方法1/3：prepareQuery查询临时表：Spark JDBC驱动对存储过程创建的临时表支持有限，临时表的生命周期绑定在存储过程执行的会话中，Spark后续查询可能无法访问到该临时表。
• 方法2：直接在query中执行存储过程：Spark的SQL接口默认期望返回结果集，若存储过程仅执行更新无返回，会因无法读取结果集而报错。

四、拆分存储过程为独立SQL语句的方案 #

如果调用存储过程始终有问题，可将存储过程逻辑拆分为独立SQL语句逐个执行：

// 执行更新最后执行时间的语句
spark.sql("""
    UPDATE dbo.LastExecTimeTable 
    SET ExecTime = GETDATE() 
    WHERE ProcName = 'YourTargetProc'
""")

// 执行存储过程中的其他查询/更新逻辑
val businessDF = spark.sql("""
    SELECT * FROM dbo.YourBusinessTable 
    WHERE Condition = 'xxx'
""")

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