PySpark中执行Snowflake非SELECT类SQL语句的合理方案咨询

问题说明 #

使用PySpark的spark.read.format('snowflake')方式执行Snowflake的CREATE TABLE ... AS SELECT（CTAS）语句时，该API仅支持返回结果集的SELECT查询，无法执行这类DDL/DML混合语句，示例代码如下：

(
    spark.read.format('snowflake') \
    .options(**sfOptions) \
    .option("query", "create table ....") \
    .load()
)

虽然可以通过调用Scala工具类的JVM方法spark.sparkContext._jvm.net.snowflake.spark.snowflake.Utils.runQuery执行语句，但该调用仅返回JavaObject id=...，无法获取执行状态、受影响行数等有用信息，实用性不足。

推荐解决方案 #

1. 使用Python Snowflake原生连接器 #

直接使用Snowflake官方提供的Python连接器执行任意SQL语句，包括CTAS，示例代码：

import snowflake.connector

# 初始化Snowflake连接
conn = snowflake.connector.connect(
    user="你的用户名",
    password="你的密码",
    account="你的账户标识",
    warehouse="你的仓库",
    database="目标数据库",
    schema="目标schema"
)

# 执行CTAS语句
cursor = conn.cursor()
try:
    cursor.execute("CREATE TABLE new_table AS SELECT * FROM existing_table")
    print(f"执行成功，受影响行数：{cursor.rowcount}")
finally:
    cursor.close()
    conn.close()

这种方式支持所有Snowflake SQL语法，能直接获取执行结果和状态，是最直观可控的方案。

2. 通过Spark SQL结合Snowflake数据源实现 #

如果希望在Spark生态内完成操作，可以先将源表加载为Spark临时视图，再通过Spark SQL执行CTAS语句写入Snowflake：

# 将Snowflake源表加载为Spark临时视图
spark.read.format("snowflake") \
    .options(**sfOptions) \
    .option("dbtable", "existing_table") \
    .load() \
    .createOrReplaceTempView("temp_source_table")

# 执行CTAS语句，将结果写入Snowflake目标表
spark.sql("""
    CREATE TABLE snowflake.目标数据库.目标schema.new_table 
    AS SELECT * FROM temp_source_table
""")

注意：此方案需要确保Spark环境已正确配置Snowflake的JDBC依赖，且当前账号拥有足够的Snowflake权限。

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