如何在Airflow UI日志中显示Snowflake Python连接器的完整查询

解决Airflow日志截断问题，获取完整原始查询 #

Airflow默认会截断过长的日志行，导致Snowflake连接器打印的完整查询被截断，可通过以下两种方式解决：

• 调整Airflow日志配置：修改Airflow的日志格式设置，取消日志内容截断。找到Airflow的配置文件（如airflow.cfg）或自定义日志配置脚本，调整日志格式化器的格式参数，移除截断限制。例如将日志格式中的%(message).1000s改为%(message)s，或增大截断长度数值。
• 手动打印完整查询：在执行SQL前，直接用logging模块打印完整的查询语句和参数（此方式打印的是带占位符的原始语句），示例代码：

import logging

sql_query = "SELECT * FROM my_table WHERE status = %s"
params = ("ACTIVE",)
# 手动打印完整查询和参数，避免Airflow日志截断
logging.info(f"准备执行查询: {sql_query}，参数: {params}")
cur.execute(sql_query, params)

获取服务器端参数替换后的最终执行查询 #

Snowflake的服务器端参数绑定是在云端完成参数替换，客户端无法直接通过连接器的cur.query获取替换后的完整语句，可通过以下两种方式实现：

1. 手动模拟参数替换（仅调试用）：
   自行实现简单的占位符替换逻辑，注意此方式仅适用于调试场景，生产环境禁止使用（存在SQL注入风险），示例代码：

def debug_format_query(query, params):
    # 适配不同类型的参数绑定格式，如%s、:1、:param_name
    if isinstance(params, tuple):
        return query % params
    elif isinstance(params, dict):
        return query % params
    return query

sql_query = "SELECT :1 FROM orders WHERE order_id = :2"
params = ("order_total", 1001)
# 生成调试用的替换后语句
debug_query = debug_format_query(sql_query, params)
logging.info(f"调试用替换后查询: {debug_query}")
cur.execute(sql_query, params)

2. 通过Snowflake查询历史获取实际执行语句：
   利用执行查询返回的sfqid（即QUERY_ID），查询Snowflake系统视图获取真实执行的完整语句，示例代码：

cur.execute(sql_query, params)
sfqid = cur.sfqid

# 查询Snowflake查询历史获取完整执行语句
get_full_query_sql = f"""
SELECT QUERY_TEXT
FROM INFORMATION_SCHEMA.QUERY_HISTORY
WHERE QUERY_ID = '{sfqid}'
"""
cur.execute(get_full_query_sql)
full_executed_query = cur.fetchone()[0]
logging.info(f"实际执行的完整查询: {full_executed_query}")

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