贡献者:@ImPerat0R_@ThinkingChen@Ray@zhongjiajie

本教程将向您介绍一些 Airflow 的基本概念、对象以及它们在编写第一个 pipline(管道)时的用法。

定义 Pipeline(管道)的例子

以下是定义一个基本 pipline(管道)的示例。如果这看起来很复杂,请不要担心,下面将逐行说明。

  1. """
  2. Airflow 教程代码位于:
  3. https://github.com/apache/airflow/blob/master/airflow/example_dags/tutorial.py
  4. """
  5. from airflow import DAG
  6. from airflow.operators.bash_operator import BashOperator
  7. from datetime import datetime, timedelta
  8. default_args = {
  9. 'owner': 'airflow',
  10. 'depends_on_past': False,
  11. 'start_date': datetime(2015, 6, 1),
  12. 'email': ['airflow@example.com'],
  13. 'email_on_failure': False,
  14. 'email_on_retry': False,
  15. 'retries': 1,
  16. 'retry_delay': timedelta(minutes=5),
  17. # 'queue': 'bash_queue',
  18. # 'pool': 'backfill',
  19. # 'priority_weight': 10,
  20. # 'end_date': datetime(2016, 1, 1),
  21. }
  22. dag = DAG('tutorial', default_args=default_args, schedule_interval=timedelta(days=1))
  23. # t1、t2 和 t3 是通过实例化 Operators 创建的任务示例
  24. t1 = BashOperator(
  25. task_id='print_date',
  26. bash_command='date',
  27. dag=dag)
  28. t2 = BashOperator(
  29. task_id='sleep',
  30. bash_command='sleep 5',
  31. retries=3,
  32. dag=dag)
  33. templated_command = """
  34. { % for i in range(5) %}
  35. echo "{{ ds }}"
  36. echo "{{ macros.ds_add(ds, 7)}}"
  37. echo "{{ params.my_param }}"
  38. { % end for %}
  39. """
  40. t3 = BashOperator(
  41. task_id='templated',
  42. bash_command=templated_command,
  43. params={'my_param': 'Parameter I passed in'},
  44. dag=dag)
  45. t2.set_upstream(t1)
  46. t3.set_upstream(t1)

这是一个 DAG 定义文件

有一件事需要考虑(一开始可能不是很直观),这个 Airflow 的 Python 脚本实际上只是一个将 DAG 的结构指定为代码的配置文件。此处定义的实际任务将在与此脚本定义的不同上下文中运行。不同的任务在不同的时间点运行在不同的 worker(工作节点)上,这意味着该脚本不能在任务之间交叉通信。请注意,为此,我们有一个名为XCom的更高级功能。

人们有时会将 DAG 定义文件视为可以进行实际数据处理的地方 - 但事实并非如此!该脚本的目的是定义 DAG 对象。它需要快速评估(秒,而不是几分钟),因为 scheduler(调度器)将定期执行它以反映更改(如果有的话)。

导入模块

一个 Airflow 的 pipeline 就是一个 Python 脚本,这个脚本的作用是为了定义 Airflow 的 DAG 对象。让我们首先导入我们需要的库。

  1. # DAG 对象; 我们将需要它来实例化一个 DAG
  2. from airflow import DAG
  3. # Operators; 我们需要利用这个对象去执行流程!
  4. from airflow.operators.bash_operator import BashOperator

默认参数

我们即将创建一个 DAG 和一些任务,我们可以选择显式地将一组参数传递给每个任务的构造函数(这可能变得多余),或者(最好地)我们可以定义一个默认参数的字典,这样我们可以在创建任务时使用它。

  1. from datetime import datetime, timedelta
  2. default_args = {
  3. 'owner': 'airflow',
  4. 'depends_on_past': False,
  5. 'start_date': datetime(2015, 6, 1),
  6. 'email': ['airflow@example.com'],
  7. 'email_on_failure': False,
  8. 'email_on_retry': False,
  9. 'retries': 1,
  10. 'retry_delay': timedelta(minutes=5),
  11. # 'queue': 'bash_queue',
  12. # 'pool': 'backfill',
  13. # 'priority_weight': 10,
  14. # 'end_date': datetime(2016, 1, 1),
  15. }

有关 BaseOperator 参数及其功能的更多信息,请参阅airflow.models.BaseOperator文档。

另外,请注意,您可以轻松定义可用于不同目的的不同参数集。一个典型的例子是在生产和开发环境之间进行不同的设置。

实例化一个 DAG

我们需要一个 DAG 对象来嵌入我们的任务。这里我们传递一个定义为dag_id的字符串,把它用作 DAG 的唯一标识符。我们还传递我们刚刚定义的默认参数字典,同时也为 DAG 定义schedule_interval,设置调度间隔为每天一次。

  1. dag = DAG(
  2. 'tutorial', default_args=default_args, schedule_interval=timedelta(days=1))

(Task)任务

在实例化 operator(执行器)时会生成任务。从一个 operator(执行器)实例化出来的对象的过程,被称为一个构造方法。第一个参数task_id充当任务的唯一标识符。

  1. t1 = BashOperator(
  2. task_id='print_date',
  3. bash_command='date',
  4. dag=dag)
  5. t2 = BashOperator(
  6. task_id='sleep',
  7. bash_command='sleep 5',
  8. retries=3,
  9. dag=dag)

注意到我们传递了一个 BaseOperator 特有的参数(bash_command)和所有的 operator 构造函数中都会有的一个参数(retries)。这比为每个构造函数传递所有的参数要简单很多。另请注意,在第二个任务中,我们使用3覆盖了默认的retries参数值。

任务参数的优先规则如下:

  1. 明确传递参数
  2. default_args字典中存在的值
  3. operator 的默认值(如果存在)

任务必须包含或继承参数task_idowner,否则 Airflow 将出现异常。

使用 Jinja 作为模版

Airflow 充分利用了Jinja Templating的强大功能,并为 pipline(管道)的作者提供了一组内置参数和 macros(宏)。Airflow 还为 pipline(管道)作者提供了自定义参数,macros(宏)和 templates(模板)的能力。

本教程几乎没有涉及在 Airflow 中使用模板进行操作的工作领域,但本节的目的是让您知道此功能的存在,让您熟悉{{ }}双花括号的用途,并指出最常见的模板变量: {{ ds }} (今天的“日期戳”)。

  1. templated_command = """
  2. { % f or i in range(5) %}
  3. echo "{{ ds }}"
  4. echo "{{ macros.ds_add(ds, 7) }}"
  5. echo "{{ params.my_param }}"
  6. { % e ndfor %}
  7. """
  8. t3 = BashOperator(
  9. task_id='templated',
  10. bash_command=templated_command,
  11. params={'my_param': 'Parameter I passed in'},
  12. dag=dag)

请注意,templated_command包含{% %}块中的代码逻辑,引用参数如{{ ds }},调用函数方式如{{ macros.ds_add(ds, 7)}},引用用户定义的参数如{{ params.my_param }}

BaseOperator中的paramshook 允许您将参数或对象的字典传递给您的模板。请花一些时间去了解my_param这个参数是如何在模板中被使用的。

文件也可以当做bash_command的参数进行传递,例如bash_command='templated_command.sh',不过这个文件的位置要在 pipeline(管道)文件的目录内(在本例中为tutorial.py)。这可能是出于多种原因,比如将脚本的逻辑和 pipeline 代码分隔开,允许在使用不同语言编写的文件中进行正确的代码突出显示,以及灵活地构建 pipeline(管道)。还可以定义您的template_searchpath,以指向 DAG 构造函数调用中的任何文件夹位置。

使用同样的 DAG 构造函数调用,可以使用user_defined_macros来定义您自己的变量。例如,将dict(foo='bar')传递给此参数允许您在模板中使用{{ foo }} 。此外,允许您指定user_defined_filters来注册自己的过滤器。例如,将dict(hello=lambda name: 'Hello %s' % name)传递给此参数可以允许您在你的模板中使用{{ 'world' | hello }}。有关自定义过滤器的更多信息,请查看Jinja 文档

有关可以在模板中引用的变量和宏的更多信息,请务必阅读部分

设置依赖关系

我们有三个不相互依赖任务,分别是t1t2t3。以下是一些可以定义它们之间依赖关系的方法:

  1. t1.set_downstream(t2)
  2. # 这意味着 t2 会在 t1 成功执行之后才会执行
  3. # 与下面这种写法相等
  4. t2.set_upstream(t1)
  5. # 位移运算符也可用于链式运算
  6. # 用于链式关系 和上面达到一样的效果
  7. t1 >> t2
  8. # 位移运算符用于上游关系中
  9. t2 << t1
  10. # 使用位移运算符能够链接
  11. # 多个依赖关系变得简洁
  12. t1 >> t2 >> t3
  13. # 任务列表也可以设置为依赖项。
  14. # 下面的这些操作都具有相同的效果:
  15. t1.set_downstream([t2, t3])
  16. t1 >> [t2, t3]
  17. [t2, t3] << t1

请注意,在执行脚本时,在 DAG 中如果存在循环或多次引用依赖项时,Airflow 会引发异常。

回顾

到此,我们有了一个非常基本的 DAG。此时,您的代码应如下所示:

  1. """
  2. Airflow 教程代码位于:
  3. https://github.com/apache/airflow/blob/master/airflow/example_dags/tutorial.py
  4. """
  5. from airflow import DAG
  6. from airflow.operators.bash_operator import BashOperator
  7. from datetime import datetime, timedelta
  8. default_args = {
  9. 'owner': 'airflow',
  10. 'depends_on_past': False,
  11. 'start_date': datetime(2015, 6, 1),
  12. 'email': ['airflow@example.com'],
  13. 'email_on_failure': False,
  14. 'email_on_retry': False,
  15. 'retries': 1,
  16. 'retry_delay': timedelta(minutes=5),
  17. # 'queue': 'bash_queue',
  18. # 'pool': 'backfill',
  19. # 'priority_weight': 10,
  20. # 'end_date': datetime(2016, 1, 1),
  21. }
  22. dag = DAG(
  23. 'tutorial', default_args=default_args, schedule_interval=timedelta(days=1))
  24. # t1, t2 and t3 are examples of tasks created by instantiating operators
  25. t1 = BashOperator(
  26. task_id='print_date',
  27. bash_command='date',
  28. dag=dag)
  29. t2 = BashOperator(
  30. task_id='sleep',
  31. bash_command='sleep 5',
  32. retries=3,
  33. dag=dag)
  34. templated_command = """
  35. { % f or i in range(5) %}
  36. echo "{{ ds }}"
  37. echo "{{ macros.ds_add(ds, 7)}}"
  38. echo "{{ params.my_param }}"
  39. { % e ndfor %}
  40. """
  41. t3 = BashOperator(
  42. task_id='templated',
  43. bash_command=templated_command,
  44. params={'my_param': 'Parameter I passed in'},
  45. dag=dag)
  46. t2.set_upstream(t1)
  47. t3.set_upstream(t1)

测试

运行脚本

是时候进行一些测试了。首先让我们确保 pipeline(管道)能够被解析。让我们保证已经将前面的几个步骤的代码保存在tutorial.py文件中,并将文件放置在airflow.cfg设置的 DAGs 文件夹中。DAGs 的默认位置是~/airflow/dags

  1. python ~/airflow/dags/tutorial.py

如果这个脚本没有报错,那就证明您的代码和您的 Airflow 环境没有特别大的问题。

命令行元数据验证

让我们运行一些命令来进一步验证这个脚本。

  1. # 打印出所有正在活跃状态的 DAGs
  2. airflow list_dags
  3. # 打印出 'tutorial' DAG 中所有的任务
  4. airflow list_tasks tutorial
  5. # 打印出 'tutorial' DAG 的任务层次结构
  6. airflow list_tasks tutorial --tree

测试实例

让我们通过在特定日期运行实际任务实例来进行测试。通过execution_date这个上下文指定日期,它会模拟 scheduler 在特定的 日期 + 时间 运行您的任务或者 dag:

  1. # 命令样式: command subcommand dag_id task_id date
  2. # 测试 print_date
  3. airflow test tutorial print_date 2015-06-01
  4. # 测试 sleep
  5. airflow test tutorial sleep 2015-06-01

现在还记得我们早些时候利用模板都做了什么?让我们通过执行这个命令看看模板会被渲染成什么样子:

  1. # 测试模版渲染
  2. airflow test tutorial templated 2015-06-01

用过运行 bash 命令,应该会显示详细的事件日志并打印结果。

请注意,airflow test命令在本地运行任务实例时,会将其日志输出到 stdout(在屏幕上),不会受依赖项影响,并且不向数据库传达状态(运行,成功,失败,…)。它只允许测试单个任务实例。

Backfill(回填)

一切看起来都运行良好,所以此时让我们运行 backfill(回填)。backfill将尊重您的依赖关系,将日志发送到文件并与数据库通信以记录状态。如果您启动了一个 web 服务,您可以跟踪它的进度。airflow webserver将启动 Web 服务器,如果您有兴趣在 backfill(回填)过程中直观地跟踪进度。

请注意,如果使用depends_on_past=True,则单个任务实例的执行将取决于前面任务实例是否成功,除了以 start_date 作为开始时间的实例(即第一个运行的 DAG 实例),他的依赖性会被忽略。

此上下文中的日期范围是start_date和可选的end_date,它们用于使用此 dag 中的任务实例填充运行计划。

  1. # 可选,在后台以 debug 模式运行 web 服务器
  2. # airflow webserver --debug &
  3. # 在时间范围内回填执行任务
  4. airflow backfill tutorial -s 2015-06-01 -e 2015-06-07

接下来做什么

就如上面这样,您已经编写,测试并 backfill(回填)了您的第一个 Airflow 的 pipeline(管道)。将您的代码合并到一个有 scheduler(调度管理器)的代码库中,这样可以启动任务并在每天执行它。

以下是您可能想要做的一些事情:

  • 深入了解用户界面 - 点击所有内容!
  • 继续阅读文档! 特别是以下部分:

    • 命令行界面
    • Operators(运营商)
    • Macros(宏)
  • 写下你的第一个 pipline(管道)!