如何通过Databricks PySpark调用Snowflake存储过程并读取返回值

前置条件 #

确保你的Databricks集群已配置好Snowflake的Spark连接器，且Spark会话已正确关联Snowflake（比如通过配置catalog或临时数据源），能直接执行Snowflake SQL语句。

调用存储过程并提取结果 #

你可以直接用spark.sql执行存储过程调用，返回的DataFrame会包含存储过程的输出。假设存储过程返回的结果列名为STATUS（值为Success或Failure），代码实现如下：

# 调用存储过程并传入参数1
df_sp = spark.sql("CALL sptest(1)")

# 提取返回的状态值
# 存储过程返回单行结果，用first()获取首行，再提取对应列
status = df_sp.first()["STATUS"]

# 验证结果
print(f"存储过程执行状态: {status}")

额外说明 #

• 如果不确定返回列的名称，可通过df_sp.columns查看DataFrame的列列表，替换成实际列名即可。
• 若存储过程返回多行或多列结果，可根据需求用collect()遍历所有行，或者用Spark的DataFrame操作进行处理。

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