Airflow中PythonOperator的定时调度和任务依赖

在 Airflow 中，使用 PythonOperator 可以创建一个基于 Python 的任务，实现定时调度和任务依赖。PythonOperator 可以执行任意的 Python 函数作为任务，并且可以在 DAG 中设置它们的依赖关系和调度时间。

下面是一个例子，演示如何使用 PythonOperator 进行定时调度和任务依赖：

首先，我们需要导入必要的库和模块：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

然后，我们创建一个空的 DAG（有向无环图），并设置默认参数：

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG('example_dag', default_args=default_args, schedule_interval=timedelta(days=1))

接下来，我们定义一些 Python 函数作为任务的实现：

def task1():
    print('This is task 1')

def task2():
    print('This is task 2')

def task3():
    print('This is task 3')

然后，我们创建三个 PythonOperator 对象，分别指定每个任务和它们的依赖关系：

task_1 = PythonOperator(
    task_id='task_1',
    python_callable=task1,
    dag=dag
)

task_2 = PythonOperator(
    task_id='task_2',
    python_callable=task2,
    dag=dag
)

task_3 = PythonOperator(
    task_id='task_3',
    python_callable=task3,
    dag=dag
)

task_1 >> task_2 >> task_3

在上述代码中，我们创建了三个 PythonOperator 对象，并分别指定了它们的任务 ID（task_id）和 Python 函数（python_callable）。然后，我们使用 '>>' 运算符指定了任务的依赖关系，即 task_1 依赖于 task_2，task_2 依赖于 task_3。

最后，我们可以将上述 DAG 保存到 Airflow 中并启动调度：

from airflow import models

models.Variable.set("example_dag", dag)

在 Airflow 的 Web UI 中，我们可以看到 example_dag 的调度和任务依赖图。根据我们定义的调度时间和依赖关系，任务将会定期执行，并按照指定的顺序执行。

总结起来，Airflow 中的 PythonOperator 可以帮助我们实现定时调度和任务依赖。通过指定任务的 Python 函数和依赖关系，我们可以灵活地管理任务的执行顺序和时间。这样，我们可以根据需要管理和调度各种类型的任务，从而更好地管理数据流程和工作流程。