如何使用Python实现Snowflake表列值与指定列表的比对校验

核心逻辑 #

• 先把待比对的下划线分隔字符串列表，统一将下划线替换为空格，得到实际要匹配的目标值
• 连接Snowflake查询目标字段的去重结果，优先用集合存储查询结果，匹配效率比列表高几个量级
• 逐个校验目标值是否存在于字段值集合中，任意值匹配失败直接终止脚本，全部校验通过才执行后续流程
• 默认对字符串做首尾空格去除处理，避免因为字段里存了不可见的首尾空格导致匹配误判，不需要可以删掉对应处理逻辑

前置依赖 #

先安装Snowflake官方Python连接器，执行命令：

pip install snowflake-connector-python

实现代码（小数据量场景，字段去重值10万条以内适用） #

import snowflake.connector
import sys

# 待比对的下划线格式字符串列表
target_list = [
    "demo1_is_good",
    "thank_you_stack",
    "what_are_you_doing"
]
# 预处理：下划线替换为空格
check_values = [item.replace("_", " ").strip() for item in target_list]

# 替换为你自己的Snowflake连接参数
sf_config = {
    "user": "你的Snowflake用户名",
    "password": "你的Snowflake密码",
    "account": "你的Snowflake账号标识（格式如xxxxxx.aws_ap_southeast_1）",
    "warehouse": "使用的计算仓库名",
    "database": "dev",
    "schema": "devschema"
}

if __name__ == "__main__":
    conn = None
    try:
        conn = snowflake.connector.connect(**sf_config)
        cur = conn.cursor()
        # 注意：Snowflake三级路径为 库.模式.表，如果原SQL里的devtable和devschema顺序写反了自行调整
        cur.execute('SELECT DISTINCT "power" FROM "dev"."devtable"."devschema"')
        # 拉取查询结果转成集合，过滤空值
        power_values = {row[0].strip() for row in cur.fetchall() if row[0] is not None}

        # 逐个校验
        for val in check_values:
            if val not in power_values:
                print(f"校验终止：值「{val}」未在目标字段中匹配到")
                sys.exit(1)
        
        print("所有值校验通过，开始执行后续流程")
        # 后续业务逻辑写在这里即可

    except Exception as e:
        print(f"执行报错：{str(e)}")
        sys.exit(1)
    finally:
        if conn:
            conn.close()

大数据量优化方案 #

如果目标表数据量很大，字段去重值超过10万条，不要把全量去重值拉到本地，直接在Snowflake侧计算差集，性能更高，替换对应查询和校验逻辑即可：

import json
# ... 省略前面的配置和连接代码
check_sql = """
WITH field_vals AS (
    SELECT DISTINCT TRIM("power") AS val FROM "dev"."devtable"."devschema"
), input_vals AS (
    SELECT TRIM(value::VARCHAR) AS val FROM TABLE(FLATTEN(input => PARSE_JSON(%s)))
)
SELECT val FROM input_vals WHERE val NOT IN (SELECT val FROM field_vals)
"""
cur.execute(check_sql, (json.dumps(check_values),))
missing_vals = [row[0] for row in cur.fetchall()]
if missing_vals:
    print(f"校验终止，未匹配到的值：{missing_vals}")
    sys.exit(1)

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