Airflow如何仅删除当前DAG Run对应的XCOM键值对？

你的删除语句仅按dag_id过滤，会删除对应DAG下所有运行实例生成的XCOM数据，与其他并行运行的DAG Run产生资源冲突。

直接修改PostgresOperator的SQL语句，通过Airflow内置的Jinja模板变量注入当前DAG Run的唯一标识，仅删除当前实例生成的XCOM：

修改后代码示例 #

delete_xcom_task = PostgresOperator(
      task_id='delete-xcom-task',
      postgres_conn_id='postgres',
      sql="delete from public.xcom where dag_id='替换为你的实际dag_id' AND run_id = '{{ run_id }}'",
      dag=curation)

• Airflow官方PostgresOperator的sql参数默认支持Jinja模板渲染，{{ run_id }}会在任务执行时自动替换为当前DAG Run的唯一ID，确保只删除本次运行生成的XCOM
• 如果你使用的是Airflow 1.x版本，xcom表没有run_id字段，可以改用execution_date作为过滤条件，SQL写法如下：

delete from public.xcom where dag_id='替换为你的实际dag_id' AND execution_date = '{{ execution_date.isoformat() }}'

• 上线前建议先将delete改为select *验证过滤结果，确认只返回当前DAG Run的XCOM数据后再执行删除操作，避免误删数据。

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