init: airflow_demo 初始化示例 DAG 集合

.idea/.gitignore

@@ -0,0 +1,10 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/
+# 已忽略包含查询文件的默认文件夹
+/queries/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/airflow_demo.iml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="SSH (sftp://hadoop@192.168.48.10:22/home/hadoop/app/venv_airflow/bin/python3.6)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PackageRequirementsSettings" />
+  <component name="PyDocumentationSettings" />
+  <component name="ReSTService" />
+  <component name="TestRunnerService" />
+</module>

args_example.py

@@ -0,0 +1,157 @@
+# 导入 DAG 类，用于定义工作流
+from airflow import DAG
+
+# 导入 Variable 类，用于获取 Airflow 中保存的全局变量
+from airflow.models import Variable
+
+# 导入 BashOperator，用于执行 Linux Shell 命令
+from airflow.operators.bash import BashOperator
+
+# 导入 PythonOperator，用于执行 Python 函数
+from airflow.operators.python import PythonOperator
+
+# 导入 datetime，用于设置 DAG 的开始日期
+from datetime import datetime
+
+
+# ==================================================
+# 第一部分：通过运行上下文获取城市参数
+# ==================================================
+
+# **context 表示接收多个关键字参数
+# Airflow 执行任务时，会把运行上下文传递给这个函数
+def print_city(**context):
+
+    # 从上下文中获取当前 DAG Run（一次工作流运行）的配置信息
+    # dag_run.conf 通常用于获取手动触发 DAG 时传入的 JSON 参数
+    # 例如：{"city": "Guangzhou"}
+
+    # 如果没有传入配置，就使用一个空字典，避免 conf 为 None
+    conf = context["dag_run"].conf or {}
+
+    # 从配置字典中获取 city 对应的值
+    # 如果没有传入 city，就使用默认值 default_city
+    city = conf.get("city", "default_city")
+
+    # 将获取到的城市打印到任务日志中
+    print(f"The city is: {city}")
+
+
+# ==================================================
+# 第二部分：获取 Airflow Variable 全局变量
+# ==================================================
+
+def use_variable():
+
+    # 从 Airflow 的 Variables 中读取名为 foo 的变量
+    # 如果 Airflow 中不存在 foo，就使用 default_value
+    foo = Variable.get("foo", default_var="default_value")
+
+    # 打印变量的值
+    print(f"The value of foo is: {foo}")
+
+
+# ==================================================
+# 第三部分：创建 DAG，也就是工作流
+# ==================================================
+
+with DAG(
+
+    # DAG 的唯一标识名称
+    dag_id="args_example",
+
+    # 每天运行一个调度周期
+    schedule_interval="@daily",
+
+    # 设置调度开始日期
+    start_date=datetime(2023, 1, 1),
+
+    # 不补跑过去未执行的调度周期
+    catchup=False,
+
+    # 设置标签，方便在 Airflow 网页中筛选
+    tags=["demo"]
+
+) as dag:
+
+    # ==================================================
+    # 第四部分：任务一——通过 context 获取城市参数
+    # ==================================================
+
+    task = PythonOperator(
+
+        # 任务的唯一标识
+        task_id="print_city_task",
+
+        # 指定任务要调用的 Python 函数
+        # 注意：传入函数名，不要写 print_city()
+        python_callable=print_city
+
+    )
+
+    # ==================================================
+    # 第五部分：任务二——通过 Jinja 模板获取城市参数
+    # ==================================================
+
+    task_bash = BashOperator(
+
+        # 任务的唯一标识
+        task_id="print_city_bash",
+
+        # BashOperator 会在执行前解析 Jinja 模板
+        # dag_run.conf 用来获取本次 DAG Run 的配置
+        # 如果没有传入 city，就使用 default_city
+        bash_command=(
+            'echo "city is '
+            '{{ dag_run.conf.get("city", "default_city") '
+            'if dag_run and dag_run.conf else "default_city" }}"'
+        )
+
+    )
+
+    # ==================================================
+    # 第六部分：任务三——获取 Airflow 全局变量
+    # ==================================================
+
+    # 全局变量（Variable）
+    task_var = PythonOperator(
+
+        # 任务的唯一标识
+        task_id="use_variable_task",
+
+        # 调用前面定义的 use_variable 函数
+        python_callable=use_variable
+
+    )
+
+    # ==================================================
+    # 第七部分：任务四——使用宏变量
+    # ==================================================
+
+    # 官方文档参考：
+    # https://airflow.apache.org/docs/apache-airflow/2.2.5/templates-ref.html
+
+    # 宏变量可以在支持模板的 Operator 参数中使用
+    # 这里使用了 ds 和 dag.dag_id 两个模板变量
+    task_ds = BashOperator(
+
+        # 任务的唯一标识
+        task_id="print_execution_date",
+
+        # {{ ds }}：本次调度运行对应的日期，格式通常为 YYYY-MM-DD
+        # {{ dag.dag_id }}：当前 DAG 的唯一标识名称
+        bash_command=(
+            'echo "Execution date is {{ ds }}, '
+            'dag_id is {{ dag.dag_id }}"'
+        )
+
+    )
+
+    # ==================================================
+    # 第八部分：定义任务之间的依赖关系
+    # ==================================================
+
+    # task 和 task_bash 是两个并行的上游任务
+    # 两者都成功后，task_var 才可以执行
+    # task_var 成功后，task_ds 才可以执行
+    [task, task_bash] >> task_var >> task_ds

branch_example.py

@@ -0,0 +1,166 @@
+# 导入 DAG 类，用于定义工作流
+from airflow import DAG
+
+# 导入 PythonOperator：执行普通的 Python 函数
+# 导入 BranchPythonOperator：根据 Python 函数的返回值选择执行分支
+from airflow.operators.python import PythonOperator, BranchPythonOperator
+
+# 导入 DummyOperator：空操作任务，用于组织任务依赖关系
+from airflow.operators.dummy import DummyOperator
+
+# 导入 TriggerRule：用于设置任务的触发规则
+from airflow.utils.trigger_rule import TriggerRule
+
+# 导入 datetime，用于设置 DAG 的开始日期
+from datetime import datetime
+
+# 导入 random，用于生成随机数
+import random
+
+
+# ==================== 1. 定义分支选择函数 ====================
+
+def choose_branch(**kwargs):
+    """
+    分支选择函数。
+
+    BranchPythonOperator 执行该函数，
+    根据函数的返回值决定执行哪个下游任务。
+
+    **kwargs：接收 Airflow 自动传入的任务上下文参数。
+    """
+
+    # 随机生成 1～3 之间的整数，包含 1 和 3
+    random_number = random.randint(1, 3)
+
+    # 将生成的随机数输出到任务日志中，方便调试
+    print(f'Random number generated: {random_number}')
+
+    # 根据随机数选择执行的分支
+    # 注意：返回值必须是下游任务的 task_id
+
+    if random_number == 1:
+        return 'task_a'  # 随机数为 1，选择 A 分支
+
+    elif random_number == 2:
+        return 'task_b'  # 随机数为 2，选择 B 分支
+
+    else:
+        return 'task_c'  # 随机数为 3，选择 C 分支
+
+
+# ==================== 2. 定义普通任务函数 ====================
+
+def print_task(task_name):
+    """
+    普通任务函数。
+
+    task_name：任务名称，用于区分不同的分支任务。
+    """
+
+    # 输出当前执行的任务名称
+    print(f'Executing {task_name}')
+
+
+# ==================== 3. 创建 DAG 工作流 ====================
+
+with DAG(
+    # DAG 的唯一标识符
+    dag_id='branch_example',
+
+    # 设置调度周期为每天执行一次
+    schedule_interval='@daily',
+
+    # 设置 DAG 开始日期
+    start_date=datetime(2026, 10, 1),
+
+    # 开启历史补跑
+    # Airflow 会根据调度规则处理开始日期之后符合条件的历史调度周期
+    catchup=True,
+
+    # 设置 DAG 标签，方便在 Airflow 页面中查找
+    tags=['demo']
+
+) as dag:
+
+    # ==================== 4. 定义分支选择任务 ====================
+
+    branching_task = BranchPythonOperator(
+        # 任务 ID
+        task_id='branching_task',
+
+        # 指定执行的 Python 函数
+        # 函数的返回值决定执行哪个下游分支
+        python_callable=choose_branch,
+    )
+
+    # ==================== 5. 定义 A 分支任务 ====================
+
+    task_a = PythonOperator(
+        # 任务 ID，必须与分支函数返回的名称一致
+        task_id='task_a',
+
+        # 使用 lambda 创建一个无参数函数
+        # 执行任务时，调用 print_task('Task A')
+        python_callable=lambda: print_task('Task A'),
+    )
+
+    # ==================== 6. 定义 B 分支任务 ====================
+
+    task_b = PythonOperator(
+        # B 分支的任务 ID
+        task_id='task_b',
+
+        # 执行时打印 Task B
+        python_callable=lambda: print_task('Task B'),
+    )
+
+    # ==================== 7. 定义 C 分支任务 ====================
+
+    task_c = PythonOperator(
+        # C 分支的任务 ID
+        task_id='task_c',
+
+        # 执行时打印 Task C
+        python_callable=lambda: print_task('Task C'),
+    )
+
+    # ==================== 8. 定义结束任务 ====================
+
+    end_task = DummyOperator(
+        # 结束任务的 ID
+        task_id='end_task',
+
+        # 设置触发规则：
+        # 所有上游任务中没有失败的任务，并且至少有一个上游任务成功，
+        # 当前任务就可以执行。
+        #
+        # 因为分支选择会把未选中的任务标记为 skipped，
+        # 所以不能简单使用默认的 all_success，
+        # 否则 end_task 可能因为存在 skipped 上游任务而被跳过。
+        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
+    )
+
+    # ==================== 9. 设置任务依赖关系 ====================
+
+    # 分支任务执行完成后，根据返回值选择一个分支：
+    #
+    # 返回 'task_a' -> 执行 task_a
+    # 返回 'task_b' -> 执行 task_b
+    # 返回 'task_c' -> 执行 task_c
+    #
+    # 未被选中的分支会被标记为 skipped。
+    #
+    # 被选中的分支执行完成后，end_task 在满足触发规则时执行。
+
+    branching_task >> [task_a, task_b, task_c] >> end_task
+
+    # 上面的依赖关系等价于：
+    #
+    # branching_task >> task_a
+    # branching_task >> task_b
+    # branching_task >> task_c
+    #
+    # task_a >> end_task
+    # task_b >> end_task
+    # task_c >> end_task

dag_dependency1.py

@@ -0,0 +1,141 @@
+# 导入 pendulum，用于创建带时区的日期时间对象
+import pendulum
+
+# 导入 DAG，用于定义工作流
+from airflow import DAG
+
+# 导入 BashOperator，用于执行 Bash 命令
+from airflow.operators.bash import BashOperator
+
+# 导入 TriggerDagRunOperator，用于触发其他 DAG 运行
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+
+# 导入 Python 的 datetime 模块
+import datetime
+
+# 导入 DagRun 模型，用于查询 DAG 的运行记录
+from airflow.models.dagrun import DagRun
+
+
+# 获取指定 DAG 最近一次运行的执行时间
+def getExecutionDate(kwargs=None):
+    # 查询 dag_id 为 dag_dependency1 的所有 DAG 运行记录
+    # find() 返回 DagRun 对象列表
+    # pop() 取出列表中的最后一个元素
+    # execution_date 获取该次运行对应的逻辑执行时间
+    return DagRun.find(dag_id="dag_dependency1").pop().execution_date
+
+
+# 定义一个 DAG，即一个工作流
+with DAG(
+    # DAG 的唯一标识，其他 DAG 可以通过这个 ID 引用它
+    dag_id='dag_dependency1',
+
+    # DAG 的描述信息
+    description="this is a hello world workflow",
+
+    # 不设置定时调度，需要手动触发或由其他 DAG 触发
+    schedule_interval=None,
+
+    # 设置 DAG 的开始时间，并指定 UTC 时区
+    # 注意：start_date 是调度的起始时间，不代表 DAG 会在这个时间自动运行
+    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
+
+    # 关闭历史任务补跑
+    # 对于配置了定时调度的 DAG，catchup=False 表示不会补跑所有历史调度周期
+    catchup=False,
+
+    # 添加标签，便于在 Airflow Web UI 中筛选 DAG
+    tags=['demo'],
+
+) as dag:
+
+    # -------------------- 第一个任务 --------------------
+    # 执行一条 Bash 命令，在日志中输出提示信息
+    first_task = BashOperator(
+        # 任务的唯一标识
+        task_id="first_task",
+
+        # 需要执行的 Bash 命令
+        bash_command="echo 'this is first task'",
+
+        # 指定该任务所属的 DAG
+        dag=dag
+    )
+
+    # -------------------- 触发第一个子 DAG --------------------
+    # 当该任务执行时，触发 args_example 这个 DAG
+    trigger_task = TriggerDagRunOperator(
+        # 当前触发任务的 ID
+        task_id="triger_task",
+
+        # 要触发的目标 DAG 的 dag_id
+        trigger_dag_id="args_example",
+
+        # 传递给子 DAG 的配置参数
+        # 子 DAG 可以通过 dag_run.conf 获取这些参数
+        # 例如：dag_run.conf.get("city") 将得到 "beijing"
+        conf={"city": "beijing"},
+
+        # 是否等待子 DAG 执行结束
+        # True 表示等待子 DAG 完成后，当前触发任务才结束
+        wait_for_completion=True,
+
+        # 每隔 5 秒检查一次子 DAG 的运行状态
+        # 该参数在等待子 DAG 完成时生效
+        poke_interval=5
+    )
+
+    # -------------------- 触发第二个子 DAG --------------------
+    # 当该任务执行时，触发 branch_example 这个 DAG
+    trigger_task2 = TriggerDagRunOperator(
+        # 当前触发任务的 ID
+        task_id="triger_task2",
+
+        # 要触发的目标 DAG 的 dag_id
+        trigger_dag_id="branch_example",
+
+        # 传递给子 DAG 的配置参数
+        # 子 DAG 可以通过 dag_run.conf 获取 city 参数
+        conf={"city": "beijing"},
+
+        # 等待子 DAG 执行结束后，再将当前任务标记为完成
+        wait_for_completion=True,
+
+        # 每隔 5 秒检查一次子 DAG 的执行状态
+        poke_interval=5
+    )
+
+    # -------------------- 最后一个任务 --------------------
+    # 当前面两个触发任务都成功完成后，执行该任务
+    last_task = BashOperator(
+        # 任务的唯一标识
+        task_id="last_task",
+
+        # 执行 Bash 命令，输出提示信息
+        bash_command="echo 'this is last task'",
+
+        # 指定该任务所属的 DAG
+        dag=dag
+    )
+
+    # -------------------- 设置任务依赖关系 --------------------
+
+    # 并行触发模式
+    # 1. first_task 先执行
+    # 2. first_task 成功后，trigger_task 和 trigger_task2 可以并行执行
+    # 3. 两个触发任务都成功完成后，last_task 才执行
+    first_task >> [trigger_task, trigger_task2] >> last_task
+
+    # -------------------- 串行触发模式 --------------------
+    # 如果希望两个子 DAG 按顺序触发，可以使用下面的依赖关系
+    # 执行顺序：first_task
+    #          -> trigger_task
+    #          -> trigger_task2
+    #          -> last_task
+    #
+    # 因为 wait_for_completion=True，
+    # trigger_task 会等待 args_example 执行完成，
+    # 然后才会执行 trigger_task2。
+    #
+    # first_task >> trigger_task >> trigger_task2 >> last_task

dag_dependency2.py

@@ -0,0 +1,112 @@
+import datetime
+import pendulum
+
+from airflow import DAG
+from airflow.sensors.external_task import ExternalTaskSensor
+from airflow.operators.bash import BashOperator
+from airflow.models import DagRun
+
+
+# 自定义函数：获取上游 DAG 最近一次查询结果中的 execution_date
+# execution_date_fn 会调用这个函数，用于确定下游需要等待哪个上游执行日期
+def getExecutionDate(kwargs=None):
+    # DagRun.find() 根据 dag_id 查询 dag_dependency1 的 DAG 运行记录
+    # pop() 取出查询结果列表中的最后一个元素，并获取该次运行的 execution_date
+    # 注意：这里取的是查询结果列表的最后一个元素，不一定能保证它就是时间上最新的一次运行
+    return DagRun.find(dag_id="dag_dependency1").pop().execution_date
+
+
+# 创建下游 DAG：dag_dependency2
+# 适用于需要等待上游 DAG 执行完成后，再执行下游任务的场景
+dag = DAG(
+    # 当前 DAG 的唯一标识
+    dag_id='dag_dependency2',
+
+    # DAG 描述信息，可在 Airflow Web UI 中查看
+    description='this is a hello world workflow',
+
+    # 不设置定时调度，None 表示该 DAG 不会按照固定时间自动调度
+    # 可以通过 Airflow Web UI、命令行或其他 DAG 手动触发
+    schedule_interval=None,
+
+    # DAG 的开始时间，使用 UTC 时区
+    # 它是调度相关的起始时间，不代表 DAG 一定会在这个时间自动运行
+    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
+
+    # 关闭历史任务补跑
+    catchup=False,
+
+    # 单次 DAG Run 的最长运行时间为 60 分钟
+    dagrun_timeout=datetime.timedelta(minutes=60),
+
+    # 添加标签，方便在 Airflow Web UI 中分类和筛选 DAG
+    tags=['demo']
+)
+
+
+# 创建外部任务传感器
+# ExternalTaskSensor 用于等待另一个 DAG 或其中的某个任务达到指定状态
+# 当前配置监控上游 DAG：dag_dependency1
+external_task = ExternalTaskSensor(
+    # 当前传感器任务的名称
+    task_id='external_task',
+
+    # 指定需要监控的上游 DAG 的 dag_id
+    external_dag_id="dag_dependency1",
+
+    # 设置为 None，表示监控整个上游 DAG 的状态
+    # 即等待上游 DAG Run 达到允许的完成状态，而不是只等待某一个指定任务
+    external_task_id=None,
+
+    # 指定需要检查的上游 DAG 执行日期
+    # 默认情况下，可通过 execution_delta 指定相对当前执行日期的偏移量
+    #
+    # 例如，execution_delta=datetime.timedelta(minutes=1)
+    # 表示检查当前执行日期之前 1 分钟对应的上游 DAG Run
+    #
+    # 注意：execution_delta 通常用于存在固定时间间隔的上、下游调度场景
+    # 当前代码注释掉该参数，因此不会使用这个时间偏移量
+    # execution_delta=datetime.timedelta(minutes=1),
+
+    # 自定义执行日期计算函数
+    # 由 getExecutionDate() 返回需要监控的上游 DAG 执行日期
+    # 当前函数直接查询 dag_dependency1 的运行记录，并取查询结果的最后一条
+    # 因此这里不是根据当前下游执行日期计算偏移量
+    execution_date_fn=getExecutionDate,
+
+    # 每隔 5 秒检查一次上游 DAG 是否达到要求的状态
+    poke_interval=5,
+
+    # 最多等待 180 秒
+    # 如果超过该时间上游 DAG 仍未达到要求的状态，传感器任务将超时失败
+    timeout=180,
+
+    # 将当前传感器任务关联到 dag_dependency2
+    dag=dag
+)
+
+
+# 创建下游 Bash 任务
+# 只有当 external_task 传感器任务成功后，该任务才会开始执行
+next_task = BashOperator(
+    # 当前任务的唯一标识
+    task_id='next_task',
+
+    # 执行一条 Shell 命令，在终端输出提示信息
+    bash_command='echo "last dag is completed"',
+
+    # 将当前任务关联到 dag_dependency2
+    dag=dag
+)
+
+
+# 设置任务依赖关系
+# external_task 成功完成后，才会执行 next_task
+#
+# 执行流程：
+# 等待 dag_dependency1 完成
+#             ↓
+#       external_task
+#             ↓
+#         next_task
+external_task >> next_task

file_sensor_example.py

@@ -0,0 +1,110 @@
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.dummy import DummyOperator
+# 导入 HiveOperator，用于执行 Hive SQL，目前暂时不使用
+# from airflow.providers.apache.hive.operators.hive import HiveOperator
+
+# 导入外部任务传感器：等待其他 DAG 中的任务完成
+from airflow.sensors.external_task import ExternalTaskSensor
+
+# 导入文件传感器：等待指定文件出现
+from airflow.sensors.filesystem import FileSensor
+
+from datetime import datetime, timedelta
+
+
+# 定义一个 DAG，即一组具有依赖关系的任务
+with DAG(
+    dag_id='file_sensor_example',       # DAG 的唯一标识
+    schedule_interval=None,              # 不设置定时调度，需要手动触发 DAG
+    start_date=datetime(2024, 10, 1),   # DAG 的开始日期
+    catchup=False,                       # 不补执行过去未运行的调度任务
+    tags=["demo"]                        # 给 DAG 添加标签，方便在 Web 页面筛选
+) as dag:
+
+    # 1. 开始任务
+    # DummyOperator 是一个空操作任务，本身不执行实际业务逻辑。
+    # 通常用于标记工作流的开始、结束，或者组织任务依赖关系。
+    start = DummyOperator(
+        task_id="start"                 # 当前任务的唯一标识
+    )
+
+    # 2. 文件传感器
+    # FileSensor 本质上也是一个 Operator。
+    # 它会反复检查指定文件是否存在，满足条件后才允许下游任务执行。
+    wait_for_file = FileSensor(
+        task_id='wait_for_file',                    # 任务 ID
+        filepath='/home/hadoop/app/airflow/success',    # 等待出现的文件路径
+        fs_conn_id='fs_default',                    # 文件系统连接 ID
+        poke_interval=5,                            # 每隔 5 秒检查一次
+        timeout=30,                                  # 最多等待 30 秒
+        mode='poke'                                 # 使用 poke 模式
+        # poke 模式下，传感器每次检查后继续占用 Worker 工作槽等待。
+        # 超过 timeout 仍未检测到文件，则任务超时失败。
+    )
+
+    # 3. 外部任务传感器
+    # ExternalTaskSensor 用来等待另一个 DAG 中的指定任务。
+    # 只有满足指定的外部任务状态条件后，当前传感器才会成功。
+    wait_for_other_task = ExternalTaskSensor(
+        task_id='wait_for_other_task',       # 当前传感器的任务 ID
+
+        external_dag_id='ods',               # 要等待的外部 DAG ID
+        external_task_id='user_ods',         # 要等待的外部任务 ID
+
+        # 当前 DAG 运行时，查找逻辑日期早 1 天的外部任务运行记录。
+        # 例如，当前逻辑日期为 10 月 12 日，
+        # 则查找逻辑日期为 10 月 11 日的 ods.user_ods。
+        execution_delta=timedelta(days=1),
+
+        allowed_states=["success"],          # 外部任务成功后，传感器才成功
+
+        poke_interval=5,                     # 每隔 5 秒检查一次外部任务状态
+        timeout=3600,                         # 最多等待 3600 秒，也就是 1 小时
+
+        # 未指定 mode 时，默认通常为 poke。
+        # 传感器会持续轮询外部任务状态。
+    )
+
+    # 4. Hive 任务示例
+    # HiveOperator 用于提交并执行 Hive SQL。
+    # 目前整段代码被注释，因此不会创建这个任务。
+    #
+    # user_dwd_task = HiveOperator(
+    #     task_id='all_finished',              # Hive 任务 ID
+    #     hql="""
+    #         insert overwrite xxx
+    #         select ....
+    #     """                                  # 需要执行的 Hive SQL
+    # )
+
+    # 5. Bash 任务
+    # BashOperator 用于执行 Linux Shell 命令。
+    some_task = BashOperator(
+        task_id='all_finished',               # 当前任务的唯一标识
+        # {{ ts }} 是 Airflow 模板变量，运行时会被替换为当前 DAG Run 的逻辑时间戳。
+        bash_command='echo "Execution time is {{ ts }}"'
+    )
+
+    # 6. 定义任务之间的依赖关系
+    # >> 表示设置执行顺序：左侧任务成功后，右侧任务才能执行。
+    #
+    # 执行顺序：
+    # start → wait_for_file → some_task
+    #
+    # 即：
+    # ① 执行 start；
+    # ② 等待指定文件出现；
+    # ③ 文件检测成功后，执行 Bash 命令。
+    start >> wait_for_file >> some_task
+
+    # 7. 外部任务依赖示例（当前已注释，不会生效）
+    # 如果需要等待 ods DAG 中的 user_ods 任务成功，
+    # 然后再执行 user_dwd_task，可以配置下面的依赖关系。
+    #
+    # start >> wait_for_other_task >> user_dwd_task
+    #
+    # 注意：
+    # 当前 user_dwd_task 的定义也被注释了，因此不能直接启用上面的依赖。
+    # 此外，ExternalTaskSensor 虽然已经定义，但当前没有连接到任何任务，
+    # 所以它不会按照这条工作流依赖链执行。

hello_world.py

@@ -0,0 +1,52 @@
+from datetime import datetime, timedelta
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.operators.bash import BashOperator
+
+# 默认参数定义
+default_args = {
+    'owner': 'xiaozhang',
+    'depends_on_past': False,
+    'email_on_failure': False,
+    "on_failure_callback":xx,
+    'retries': 1,
+    'retry_delay': timedelta(minutes=5),
+}
+
+# dag = DAG(
+#     dag_id='hello_world',
+#     default_args=default_args,
+#     start_date=datetime(2023, 1, 1),
+#     schedule_interval='@daily',
+#     catchup=False,
+#     tags= ["demo"]
+# )
+
+# 定义任务
+def print_hello():
+    print("Hello, World!")
+
+# 实例化DAG
+with DAG(
+    dag_id='hello_world',
+    default_args=default_args,
+    description="这个是一个样例",
+    start_date=datetime(2024, 10, 1),
+    schedule_interval='@daily',
+    catchup=True,
+    tags= ["demo","xiaozhang"]
+) as dag:
+
+    task1 = PythonOperator(
+        task_id='print_hello',
+        python_callable=print_hello
+    )
+
+    task2 = BashOperator(
+        task_id='run_bash_command',
+        bash_command='echo "This is a Bash command"',
+    )
+
+    # 定义依赖关系
+    task1 >> task2  # task2将在task1完成后执行
+    # task1.set_downstream(task2) # 这种写法也可以，但是不太推荐，因为不够直观

hello_world2.py

@@ -0,0 +1,59 @@
+from datetime import datetime, timedelta
+
+# 导入 Airflow 所需的类
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.operators.bash import BashOperator
+
+
+# 定义任务最终失败时的回调函数
+def task_failure_callback(context):
+    # 获取当前任务实例
+    task_instance = context["task_instance"]
+
+    # 打印失败任务的名称和所属 DAG 名称
+    print(f"任务执行失败：{task_instance.task_id}")
+    print(f"DAG 名称：{task_instance.dag_id}")
+
+
+# 定义任务的默认参数
+default_args = {
+    "owner": "xiaozhang",  # 任务负责人
+    "depends_on_past": False,  # 不要求依赖上一次运行的结果
+    "email_on_failure": False,  # 关闭邮件失败通知
+    "on_failure_callback": task_failure_callback,  # 任务最终失败时调用
+    "retries": 1,  # 失败后最多额外重试 1 次
+    "retry_delay": timedelta(minutes=5),  # 重试前等待 5 分钟
+}
+
+
+# 定义一个普通的 Python 函数
+def print_hello():
+    print("Hello, World!")
+
+
+# 创建 DAG，也就是工作流
+with DAG(
+    dag_id="hello_world2",  # 改成新的 DAG ID
+    default_args=default_args,
+    description="这个是一个样例",
+    start_date=datetime(2024, 10, 1),
+    schedule_interval="@daily",
+    catchup=False,
+    tags=["demo", "xiaozhang"],
+) as dag:
+
+    # 任务一：执行 Python 函数
+    task1 = PythonOperator(
+        task_id="print_hello",  # 任务 ID
+        python_callable=print_hello,  # 传入函数本身，不加括号
+    )
+
+    # 任务二：执行 Bash 命令
+    task2 = BashOperator(
+        task_id="run_bash_command",  # 任务 ID
+        bash_command='echo "This is a Bash command"',  # 要执行的命令
+    )
+
+    # 设置依赖：task1 成功后，task2 才会执行
+    task1 >> task2

hive_provider.py

@@ -0,0 +1,46 @@
+
+import pendulum
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.dummy import DummyOperator
+from airflow.providers.apache.hive.operators.hive import HiveOperator
+
+with DAG(
+    dag_id='hive_provider',
+    schedule_interval=None,
+    start_date=pendulum.datetime(2024, 10, 1, tz="UTC"),
+    # 如果是True，配置定时运行的时候，它会补足所有缺少的任务
+    catchup=False,
+    tags=['demo'],
+) as dag:
+    start = DummyOperator(
+        task_id="hive_start"
+    )
+
+    # 执行以下hql
+    hive_operator = HiveOperator(
+        task_id="hive_operator",
+        hql="select * from test_db.person"
+    )
+
+    hive_operator2 = HiveOperator(
+        task_id="hive_operator2",
+        hql="select * from test_db.person limit 1"
+    )
+
+    hive_operator3 = HiveOperator(
+        task_id="hive_operator3",
+        hql="select * from test_db.person",
+        hive_cli_conn_id="hiveserver2_default",
+    )
+
+    hive_post_action = BashOperator(
+        task_id="hive_post_action",
+        bash_command="echo 'last hql had been executed'",
+        dag=dag
+    )
+
+    # start >> hive_operator >> hive_post_action
+    # start >> hive_operator2 >> hive_post_action
+    start >> [hive_operator, hive_operator2,hive_operator3] >> hive_post_action

xcom_example.py

@@ -0,0 +1,100 @@
+# 导入 DAG 类，用于定义工作流
+from airflow import DAG
+
+# 导入 PythonOperator，用于执行 Python 函数
+from airflow.operators.python import PythonOperator
+
+# 导入 datetime，用于设置 DAG 的开始日期
+from datetime import datetime
+
+
+def push_data(**context):
+    """
+    向 xcom 推送数据的任务。
+
+    **context：接收 Airflow 传入的上下文参数。
+    其中包含 ti（TaskInstance），可以通过 ti 操作 xcom。
+    """
+
+    # 获取当前任务实例，并向 xcom 写入一条数据
+    # key='my_key'：xcom 数据的键
+    # value='my_value'：实际存储的数据
+    context['ti'].xcom_push(
+        key='my_key',
+        value='my_value'
+    )
+
+    # 返回一个字符串
+    # PythonOperator 默认会将函数返回值自动保存到 xcom
+    # 默认的 xcom key 是 'return_value'
+    return "默认的push"
+
+
+def pull_data(**context):
+    """
+    从上游任务的 xcom 中读取数据。
+    """
+
+    # 获取当前任务实例（TaskInstance）
+    ti = context['ti']
+
+    # 从 push_task 任务中读取 key 为 my_key 的数据
+    # 这里读取的是之前通过 xcom_push() 手动推送的数据
+    value = ti.xcom_pull(
+        task_ids='push_task',
+        key='my_key'
+    )
+
+    # 从 push_task 任务中读取默认的返回值
+    # PythonOperator 会将 push_data() 的返回值
+    # 自动保存到 key='return_value' 对应的 xcom 中
+    return_value = ti.xcom_pull(
+        task_ids='push_task',
+        key='return_value'
+    )
+
+    # 将获取到的数据输出到任务日志中，方便查看执行结果
+    print(f'Pulled value: {value}')
+    print(f'return_value: {return_value}')
+
+
+# 定义一个 DAG，即 Airflow 工作流
+with DAG(
+    # DAG 的唯一标识符
+    dag_id='xcom_example',
+
+    # DAG 的开始日期
+    # 注意：开始日期不代表 DAG 会在这一天自动执行
+    start_date=datetime(2024, 10, 1),
+
+    # 不设置定时调度，通常需要手动触发 DAG
+    schedule_interval=None,
+
+    # 不补执行历史上未运行的调度任务
+    catchup=False,
+
+    # 为 DAG 添加标签，方便在 Airflow 页面中筛选
+    tags=["demo"]
+) as dag:
+
+    # 定义第一个任务：向 xcom 写入数据
+    push_task = PythonOperator(
+        # 任务的唯一标识符
+        task_id='push_task',
+
+        # 指定需要执行的 Python 函数
+        python_callable=push_data
+    )
+
+    # 定义第二个任务：从 xcom 读取数据
+    pull_task = PythonOperator(
+        # 任务的唯一标识符
+        task_id='pull_task',
+
+        # 指定需要执行的 Python 函数
+        python_callable=pull_data
+    )
+
+    # 设置任务依赖关系：
+    # push_task 必须先成功完成，pull_task 才会执行
+    push_task >> pull_task

xcom_example2.py

@@ -0,0 +1,145 @@
+# 导入 sleep，用于模拟模型训练所需的时间
+from time import sleep
+
+# 导入 DAG，用于定义 Airflow 工作流
+from airflow import DAG
+
+# 导入 PythonOperator，用于执行 Python 函数
+from airflow.operators.python import PythonOperator
+
+# 导入 datetime，用于设置 DAG 的开始日期
+from datetime import datetime
+
+# 导入 random，用于生成随机准确率
+import random
+
+
+# ==================== 1. 模拟模型训练任务 ====================
+def train_model(model_name, **context):
+    """
+    模拟训练一个机器学习模型，并将模型准确率存入 XCom。
+
+    model_name：模型名称，例如 model_A、model_B、model_C
+    **context：Airflow 传入的运行上下文，其中包含 ti 等信息
+    """
+
+    # 模拟模型训练过程，让当前任务等待 10 秒
+    sleep(10)
+
+    # 随机生成一个 0.7 到 1.0 之间的浮点数，模拟模型准确率
+    accuracy = random.uniform(0.7, 1.0)
+
+    # 将模型准确率手动推送到 XCom
+    # key='model_accuracy'：这条数据在 XCom 中的键名
+    # value=accuracy：实际保存的准确率
+    context['ti'].xcom_push(
+        key='model_accuracy',
+        value=accuracy
+    )
+
+    # 将模型名称和准确率输出到任务日志中
+    print(f'{model_name} accuracy: {accuracy}')
+
+    # 注意：这里没有 return accuracy
+    # 因此不会把 accuracy 作为函数返回值保存到默认的 return_value 中
+
+
+# ==================== 2. 选择最佳模型 ====================
+def choose_best_model(**context):
+    """
+    从三个训练任务的 XCom 中读取准确率，并找出准确率最高的模型。
+    """
+
+    # 获取当前任务实例 TaskInstance
+    ti = context['ti']
+
+    # 一次性读取三个上游任务的准确率
+    accuracies = ti.xcom_pull(
+        task_ids=['model_A', 'model_B', 'model_C'],  # 指定要读取的任务
+        key='model_accuracy'                         # 指定要读取的 XCom 键
+    )
+
+    # 输出读取到的准确率列表
+    # 例如：[0.82, 0.95, 0.88]
+    print(f'accuracies: {accuracies}')
+
+    # 找出列表中的最大准确率
+    # max(accuracies) 返回最大值，例如 0.95
+    # accuracies.index(...) 返回该值第一次出现的索引，例如 1
+    best_model_index = accuracies.index(max(accuracies))
+
+    # 输出最高准确率
+    # 注意：这里原代码通过索引拼出了 model_1、model_2 或 model_3，
+    # 并不是实际任务 ID model_A、model_B 或 model_C。
+    print(
+        f'Best model accuracy: {accuracies[best_model_index]} '
+        f'from model_{best_model_index + 1}'
+    )
+
+
+# ==================== 3. 定义 DAG 工作流 ====================
+with DAG(
+    # DAG 的唯一标识符
+    dag_id='xcom_example2',
+
+    # DAG 的开始日期；不代表当天一定会自动执行
+    start_date=datetime(2024, 10, 1),
+
+    # 不设置定时调度，通常需要手动触发 DAG
+    schedule_interval=None,
+
+    # 不补跑历史上未执行的调度任务
+    catchup=False,
+
+    # 为 DAG 添加标签，便于在 Airflow 页面中筛选
+    tags=["demo"]
+) as dag:
+
+    # ==================== 4. 定义模型 A 训练任务 ====================
+    model_A = PythonOperator(
+        # Airflow 中的任务 ID
+        task_id='model_A',
+
+        # 指定要执行的 Python 函数
+        python_callable=train_model,
+
+        # 向 train_model 函数传递额外参数
+        # 相当于调用 train_model(model_name='model_A', ...)
+        op_kwargs={'model_name': 'model_A'}
+    )
+
+    # ==================== 5. 定义模型 B 训练任务 ====================
+    model_B = PythonOperator(
+        task_id='model_B',
+        python_callable=train_model,
+        op_kwargs={'model_name': 'model_B'}
+    )
+
+    # ==================== 6. 定义模型 C 训练任务 ====================
+    model_C = PythonOperator(
+        task_id='model_C',
+        python_callable=train_model,
+        op_kwargs={'model_name': 'model_C'}
+    )
+
+    # ==================== 7. 定义最佳模型选择任务 ====================
+    choose_task = PythonOperator(
+        # 任务 ID
+        task_id='choose_best_model',
+
+        # 指定执行选择最佳模型的函数
+        python_callable=choose_best_model
+    )
+
+    # ==================== 8. 设置任务依赖关系 ====================
+
+    # 三个模型任务互相没有依赖关系，可以并行执行
+    # choose_task 必须等待三个模型任务全部成功完成后才会执行
+    [model_A, model_B, model_C] >> choose_task
+
+
+# 脚本被直接执行时，调用 DAG 的 CLI 入口
+# 这是旧式的命令行调试写法，不是部署 DAG 必需的内容；
+# 正常使用 Airflow 调度器时，通常可以删除这段代码。
+if __name__ == '__main__':
+    dag.cli()