如何从Spark调用Snowflake存储过程？已完成连接但调用失败

问题根源 #

spark.read 是用于读取数据生成DataFrame的API，要求执行的SQL语句必须返回可被Spark解析的标准结果集。如果你的存储过程无返回结果、或返回的不是标准结果集，用这种方式调用就会失败；另外原代码里的query参数值未加引号，属于语法错误。

解决方案 #

1. 执行无返回结果的存储过程

通过Spark底层JDBC连接直接执行，无需生成DataFrame：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
# 建立JDBC连接
conn = spark._sc._gateway.jvm.java.sql.DriverManager.getConnection(
    sfOptions["sfUrl"],
    sfOptions["sfUser"],
    sfOptions["sfPassword"]
)
stmt = conn.createStatement()
# 执行存储过程
stmt.execute("CALL 你的存储过程名(参数1, 参数2)")
# 关闭资源
stmt.close()
conn.close()

2. 调用返回结果集的存储过程

先确保存储过程返回标准结果集，同时修正代码语法错误（给query参数的SQL语句加引号）：

df = spark.read.format(SNOWFLAKE_SOURCE_NAME) \
  .options(**sfOptions) \
  .option("query", "CALL 你的存储过程名(参数1, 参数2)") \
  .load()
# 查看返回结果
df.show()

3. 用Spark SQL直接调用

先注册Snowflake临时视图关联数据源，再执行存储过程：

# 注册任意存在的Snowflake表为临时视图（仅用于绑定数据源）
spark.read.format(SNOWFLAKE_SOURCE_NAME).options(**sfOptions).load("任意存在的表名").createOrReplaceTempView("snowflake_temp")
# 执行存储过程并查看结果
spark.sql("CALL 你的存储过程名(参数1, 参数2)").show()

注意事项 #

• 确认Snowflake账号拥有该存储过程的EXECUTE权限
• 若存储过程仅返回单个值或无返回，必须使用JDBC直接执行的方式

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