Airflow -按不带时间的执行日期查找特定dag id的dag运行

基础概念

Apache Airflow 是一个用于创建、调度和监控工作流的开源平台。它使用有向无环图（DAG）来表示工作流，每个 DAG 包含多个任务（Task），这些任务按照依赖关系依次执行。

相关优势

1. 灵活性：Airflow 允许用户定义复杂的工作流，支持任务依赖、重试机制等。
2. 可视化：提供 Web UI 来监控和管理 DAG 和任务的状态。
3. 可扩展性：支持自定义操作符、传感器和钩子，方便集成各种数据源和处理逻辑。
4. 调度：内置强大的调度器，可以按需调度任务。

类型

Airflow 中的 DAG 运行可以分为以下几种类型：

• Scheduled：按照预定的时间间隔运行。
• Backfill：用于填充历史数据，通常用于重新运行过去的数据处理任务。
• One-time：一次性运行，不按时间调度。

应用场景

Airflow 广泛应用于数据处理、ETL（Extract, Transform, Load）、机器学习工作流、批处理任务等场景。

查找特定 DAG ID 的 DAG 运行

假设你想查找不带时间的执行日期（即 execution_date 为 None）的特定 DAG ID 的 DAG 运行，可以使用 Airflow 的 API 或者直接查询数据库。

使用 Airflow API

from airflow.api.client.local_client import LocalClient

client = LocalClient()
dag_id = 'your_dag_id'

# 获取所有 DAG 运行
all_runs = client.get_dag_runs(dag_id=dag_id)

# 过滤出不带时间的执行日期的 DAG 运行
runs_without_time = [run for run in all_runs if run.execution_date is None]

for run in runs_without_time:
    print(f"DAG Run ID: {run.id}, Execution Date: {run.execution_date}")

直接查询数据库

如果你有权限直接访问 Airflow 的数据库，可以使用 SQL 查询来查找特定 DAG ID 的 DAG 运行。

SELECT id, execution_date
FROM dag_run
WHERE dag_id = 'your_dag_id' AND execution_date IS NULL;

可能遇到的问题及解决方法

问题：找不到特定 DAG ID 的 DAG 运行

原因：

1. DAG ID 错误。
2. 查询的时间范围不对。
3. 数据库中没有相应的记录。

解决方法：

1. 确认 DAG ID 是否正确。
2. 检查查询的时间范围是否正确。
3. 确认数据库中是否有相应的记录，可以手动插入一条记录进行测试。

问题：执行日期为 None 的 DAG 运行不存在

原因：

1. 该 DAG 没有被设置为一次性运行。
2. 该 DAG 的执行日期已经被设置。

解决方法：

1. 确认 DAG 的调度配置，确保它支持一次性运行。
2. 如果需要，可以手动设置执行日期为 None 并重新调度 DAG。

参考链接

• Apache Airflow Documentation
• Airflow API Documentation
• Airflow Database Schema

希望这些信息对你有所帮助！