Airflow -从BigQuery动态生成任务，但任务在之前完成之前重复运行

基础概念

Apache Airflow 是一个用于创建、调度和监控工作流的开源平台。它允许你定义工作流为有向无环图（DAG），并提供了丰富的操作符来执行各种任务，如 Bash 命令、Python 函数、数据库操作等。BigQuery 是 Google 提供的一个完全托管的、可扩展的、服务器less的数据仓库，用于大规模数据集的分析。

动态生成任务

动态生成任务意味着任务的创建不是预先定义的，而是在运行时根据某些条件或数据生成的。在 Airflow 中，这通常通过 PythonOperator 实现，它允许你编写 Python 代码来动态创建任务。

问题描述

当你在 Airflow 中从 BigQuery 动态生成任务时，可能会遇到任务在完成之前重复运行的问题。这通常是由于以下几个原因造成的：

任务状态未正确更新：Airflow 可能没有正确地记录任务的完成状态，导致任务被错误地重新调度。
并发问题：如果有多个调度器实例同时运行，可能会导致任务被重复执行。
依赖关系配置错误：任务之间的依赖关系配置不正确，导致任务在不应该执行的时候被触发。

解决方案

1. 确保任务状态正确更新

确保 Airflow 的数据库中正确记录了任务的完成状态。你可以通过以下方式检查和修复：

from airflow.models import TaskInstance

# 检查任务实例的状态
ti = TaskInstance(task_id='your_task_id', execution_date='your_execution_date')
print(ti.state)

如果状态不正确，可以尝试手动更新状态：

from airflow.utils.state import State

ti.state = State.SUCCESS
ti.update_state()

2. 避免并发问题

确保只有一个调度器实例在运行。你可以在 Airflow 的配置文件中设置 single_instance=True：

# airflow.cfg
[scheduler]
single_instance = True

3. 正确配置依赖关系

确保任务之间的依赖关系配置正确。例如，如果你有两个任务 A 和 B，B 依赖于 A 的完成，你可以这样配置：

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

with DAG('example_dag', start_date=datetime(2023, 1, 1)) as dag:
    task_a = DummyOperator(task_id='task_a')
    task_b = DummyOperator(task_id='task_b')

    task_b.set_upstream(task_a)

4. 使用 BigQuery 连接器

确保你正确配置了 Airflow 和 BigQuery 的连接。你可以在 Airflow 的 Web UI 中添加一个 BigQuery 连接：

进入 Airflow Web UI。
导航到 Admin -> Connections。
添加一个新的连接，选择 BigQuery 类型，并填写相关信息。

示例代码

以下是一个简单的示例，展示如何在 Airflow 中从 BigQuery 动态生成任务：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago
from google.cloud import bigquery

def generate_tasks():
    client = bigquery.Client()
    query = """
        SELECT id
        FROM your_dataset.your_table
        WHERE status = 'pending'
    """
    results = client.query(query).result()
    
    tasks = []
    for row in results:
        task_id = f"task_{row.id}"
        task = PythonOperator(
            task_id=task_id,
            python_callable=your_task_function,
            op_args=[row.id],
            dag=dag
        )
        tasks.append(task)
    
    return tasks

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

dag = DAG('dynamic_tasks_dag', default_args=default_args, schedule_interval='@daily')

tasks = generate_tasks()