Airflow执行Redshift INSERT INTO SELECT无数据插入问题求助

问题背景 #

有一个包含5个串行任务的Airflow DAG，其中Task4执行INSERT INTO target_table_1 SELECT * FROM view_2时出现异常：

• Airflow任务和Redshift查询历史均显示执行成功
• view_2有返回数据，且SQL客户端执行同语句可成功插入
• target_table_1表结构与view_2输出匹配，view_2过滤条件也无问题，但Task4执行后表为空

DAG任务明细：

1. 截断临时表：TRUNCATE temp_table;
2. 从view_1加载数据到temp_table：INSERT INTO temp_table SELECT * FROM view_1
3. 从temp_table加载到fact_table_1：INSERT INTO fact_table_1 SELECT * FROM temp_table
4. 从view_2加载到target_table_1：INSERT INTO target_table_1 SELECT * FROM view_2
5. 从view_3加载到target_table_2：INSERT INTO target_table_2 SELECT * FROM view_3

进一步排查建议 #

1. 检查Airflow任务的事务行为 #

• 确认Airflow的Redshift算子（如RedshiftSQLOperator）是否开启自动提交：部分算子默认不会自动提交事务，若Task4的INSERT在未提交事务中执行，Redshift查询历史会显示成功，但数据不会落地。可在SQL末尾显式添加COMMIT;，或检查算子的autocommit参数是否设为True。
• 查询Redshift的STL_TRANSACTION系统表，查看Task4对应事务的最终状态，确认是否存在隐性回滚（如内存不足、锁等待超时但未抛出异常）。

2. 验证Airflow执行SQL与本地客户端的一致性 #

• 开启Airflow任务的详细日志，查看实际发送到Redshift的SQL文本，确认是否存在模板渲染错误（比如日期变量替换异常，导致过滤条件意外变更）。
• 通过Redshift的STL_QUERY表查询Task4对应的querytxt字段，和本地执行的语句逐行比对，排查是否存在空格、别名、隐式转换等细微差异。

3. 检查执行权限与环境的差异 #

• 对比Airflow连接Redshift的账号和本地SQL客户端账号的权限：确认Airflow账号对target_table_1有INSERT权限，且不存在行级权限（RLS）限制导致插入数据被过滤。
• 检查账号的search_path设置：不同账号的默认schema可能不同，确认Airflow操作的是你查询的目标schema下的target_table_1，而非其他schema的同名表。

4. 排查Redshift隐性数据问题 #

• 查询Redshift的STL_LOAD_ERROR和STL_ERROR系统表：即使查询显示成功，也可能存在隐性数据转换错误（如字符截断、类型不匹配但未触发异常），导致插入行被丢弃。
• 修改Task4的SQL语句，添加错误日志：INSERT INTO target_table_1 SELECT * FROM view_2 LOG ERRORS INTO error_table;，之后查询error_table查看是否有被拒绝的行。

5. 核对执行时区与查询计划 #

• 确认Airflow任务的执行时区：若view_2的过滤条件依赖时区，Airflow的时区设置可能和本地客户端不同，导致实际查询出的数据为空（需注意本地查询和Airflow执行的时间上下文差异）。
• 在Airflow中执行EXPLAIN INSERT INTO target_table_1 SELECT * FROM view_2，对比本地执行的查询计划，排查是否存在执行路径差异（如索引使用、数据分布策略不同）。

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