commit 69a1df7aa0978cba61f8685cd243da73fefd535a Author: KennethCheng Date: Tue Oct 13 14:14:04 2026 +0800 init: airflow_demo 初始化示例 DAG 集合 diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..f6906f2 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,10 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ +# 已忽略包含查询文件的默认文件夹 +/queries/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/.idea/airflow_demo.iml b/.idea/airflow_demo.iml new file mode 100644 index 0000000..46c627b --- /dev/null +++ b/.idea/airflow_demo.iml @@ -0,0 +1,14 @@ + + + + + + + + + + + + + + \ No newline at end of file diff --git a/args_example.py b/args_example.py new file mode 100644 index 0000000..51a5920 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/branch_example.py b/branch_example.py new file mode 100644 index 0000000..7c5f432 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/dag_dependency1.py b/dag_dependency1.py new file mode 100644 index 0000000..c995305 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/dag_dependency2.py b/dag_dependency2.py new file mode 100644 index 0000000..581c01d --- /dev/null +++ b/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 \ No newline at end of file diff --git a/file_sensor_example.py b/file_sensor_example.py new file mode 100644 index 0000000..df28334 --- /dev/null +++ b/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 虽然已经定义，但当前没有连接到任何任务， + # 所以它不会按照这条工作流依赖链执行。 \ No newline at end of file diff --git a/hello_world.py b/hello_world.py new file mode 100644 index 0000000..140f3c4 --- /dev/null +++ b/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) # 这种写法也可以，但是不太推荐，因为不够直观 \ No newline at end of file diff --git a/hello_world2.py b/hello_world2.py new file mode 100644 index 0000000..1979849 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/hive_provider.py b/hive_provider.py new file mode 100644 index 0000000..95f7c4f --- /dev/null +++ b/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 diff --git a/xcom_example.py b/xcom_example.py new file mode 100644 index 0000000..82a5b3c --- /dev/null +++ b/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 \ No newline at end of file diff --git a/xcom_example2.py b/xcom_example2.py new file mode 100644 index 0000000..6275a64 --- /dev/null +++ b/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() \ No newline at end of file