Apache Airflow 快速入门教程

Apache Airflow已经成为Python生态体系中管道编排的究竟上的库。与类似的办理方案相反，由于它的简朴性和可扩展性，它已经得到了遍及。在本文中，我将实验概述它的告急概念，并让您清楚地相识何时以及怎样使用它。
Airflow应用场景

想象一下，你想要构建一个呆板学习管道，它由以下几个步调构成：

• 从基于云的存储中读取图像数据集
  
• 处置处罚图像
  
• 使用下载的图像练习深度学习模子
  
• 将练习好的模子上传到云端
  
• 摆设模子
  

你将怎样安排和自动化这个工作流程？Cron作业是一个简朴的办理方案，但它也带来了很多标题。最告急的是，它们不允许你有用地扩展。Airflow提供了轻松调治和扩展复杂数据流程编排的本领，另一方面，它还可以大概在故障后自动重新运行它们，管理它们的依靠关系，并使用日记和仪表板监督它们。
在构建上述数据流之前，让我们先相识Apache Airflow 的根本概念。
Airflow 简介

Apache Airflow 是一个开源的平台，用于编排、调治和监控工作流，工作流是由一系列使命（Tasks）构成的，这些使命可以是数据处置处罚、数据分析、呆板学习模子练习、文件传输等各种使用。因此，它是ETL和MLOps用例的抱负办理方案。示例用例包罗：

• 从多个数据源提取数据，对其举行聚合、转换，并将其存储在数据堆栈中。
  
• 从数据中提取见解并将其体如今分析仪表板中
  
• 练习、验证和摆设呆板学习模子
  

核心组件

在默认版本中安装Apache Airflow 时，你将看到四个差别的组件。

• Webserver: Webserver是Airflow的用户界面（UI），它允许您在不必要CLI或API的环境下与之交互。从那边可以实行和监督管道，创建与外部体系的毗连，查抄它们的数据集等等。
  
• 实行器：实行器是管道运行的机制。有很多差别范例的管道在当地运行，在单个呆板中运行，大概以分布式方式运行。一些例子是LocalExecutor， SequentialExecutor， CeleryExecutor和KubernetesExecutor
  
• 调治器：调治器负责在准确的时间实行差别的使命，重新运行管道，回填数据，确保使命完成等。
  
• PostgreSQL：存储全部管道元数据的数据库。这通常是Postgres，但也支持其他SQL数据库。
  

安装Airflow最简朴的方法是使用docker compose。你可以从这里下载官方的docker撰写文件：

1. $ curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.1/docker-compose.yaml'

复制代码

根本概念

要学习Apache Airflow，必须熟悉它的告急概念，这些概念大概有点难明白，让我们试着揭开它们的秘密面纱。
DAGs

全部管道都界说为有向无环图（dag）。每次实行DAG时，都会创建一个单独的运行。每个DAG运行都是独立的，而且包罗一个关于DAG实行阶段的状态。这意味着类似的dag可以并行实行多次。
要实例化DAG，可以使用DAG函数或与上下文管理器一起使用，如下所示：

1. from airflow import DAG
2. with DAG(
3.     "mlops",
4.     default_args={
5.         "retries": 1,
6.      },
7.     schedule=timedelta(days=1),
8.     start_date=datetime(2023, 1, 1)
9. ) as dag:
11. # dag code goes here

复制代码

上下文管理器担当一些关于DAG的全局变量和一些默认参数。默认参数被通报到全部使掷中，而且可以在每个使命的底子上重写。完备的参数列表可以在官方文档中找到。
在本例中，我们界说DAG将从2023年1月1日开始，而且每天实行一次。retries参数确保在大概出现故障后重新运行一次。
task（使命）

DAG的每个节点体现一个Task，即一段单独的代码。每个使命大概有一些上游和鄙俚依靠项。这些依靠关系体现使命怎样相互关联以及它们应该以何种序次实行。每当初始化一个新的DAG运行时，全部使命都初始化为Task实例。这意味着每个Task实例都是给定使命的特定运行。

operator(使命模板)

使用符可以被视为预界说使命的模板，由于它们封装了样板代码并抽象了它们的大部分逻辑。常见的使用符有BashOperator、PythonOperator、MySqlOperator、S3FileTransformOperator。我们看到，使用符可以界说遵照特定模式的使命。比方，MySqlOperator创建使命来实行SQL查询，而BashOperator实行bash脚本。
使用符在DAG上下文管理器中界说，如下所示。下面的代码创建了两个使命，一个实行bash下令，另一个实行MySQL查询。

1. with DAG(
2.         "tutorial"
3. ) as dag:
5.     task1 = BashOperator(
6.         task_id="print_date",
7.         bash_command="date",
8.     )
10.     task2 = MySqlOperator(
11.         task_id="load_table",
12.         sql="/scripts/load_table.sql"
13.     )

复制代码

使命依靠

为了形成DAG的布局，我们必要界说每个使命之间的依靠关系。一种方法是使用>>符号，如下所示：

1. task1 >> task2 >> task3
2. # 一个任务有多个依赖
3. task1 >> [task2, task3]
4. # 也可以使用set_downstream, set_upstream
5. t1.set_downstream([t2, t3])

复制代码

xcom

xcom，或相互通讯，负责使命之间的通讯。xcom对象可以在使命之间推拉数据。更具体地说，它们将数据推入元数据数据库，其他使命可以从中提取数据。这就是为什么可以通过它们通报的数据量是有限的。但是，如果必要传输大数据，则可以使用符合的外部数据存储，比方对象存储或NoSQL数据库。
看看下面的代码。这两个使命使用ti参数（使命实例的缩写）通过xcom举行通讯。train_model使命将model_path推入元数据数据库，元数据由deploy_model使命拉出。

1. dag = DAG(
2.     'mlops_dag',
3. )
5. def train_model(ti):
6.     model_path = train_and_save_model()
7.     ti.xcom_push(key='model_path', value=model_path)
9. def deploy_model(ti):
10.     model_path = ti.xcom_pull(key='model_path', task_ids='train_model')
11.     deploy_trained_model(model_path)
13. train_model_task = PythonOperator(
14.     task_id='train_model',
15.     python_callable=train_model,
16.     dag=dag
17. )
19. deploy_model_task = PythonOperator(
20.     task_id='deploy_model',
21.     python_callable=deploy_model,
22.     dag=dag
23. )
25. train_model_task >> deploy_model_task

复制代码

Taskflow

Taskflow API是一种使用Python装饰器@task来界说使命的简朴方法。如果全部使命的逻辑都可以用Python编写，那么一个简朴的解释就可以界说一个新使命。Taskflow自动管理其他使命之间的依靠关系和通讯。
使用Taskflow API，我们可以用@dag装饰器初始化DAG。下面是使用Tashflow示例：

1. @dag(
2.     start_date=datetime(2023, 1, 1),
3.     schedule_interval='@daily'
4. )
5. def mlops():
7.     @task
8.     def load_data():
9.         . . .
10.         return df
12.     @task
13.     def preprocessing(data):
14.        . . .
15.        return data
17.     @task
18.     def fit(data):
19.         return None
21.     df = load_data()
22.     data = preprocessing(df)
23.     model = fit(data)
24.    
25. dag = mlops()

复制代码

留意，使命之间的依靠关系是通过每个函数参数隐含的。这里我们是简朴的毗连序次，但现实可以变得复杂得多。Taskflow API还办理了使命之间的通讯标题，因此使用xcom的需求有限。
调治

作业调治是Airflow的核心功能之一。这可以使用schedule_interval参数完成，该参数吸收cron表达式，体现日期时间对象，或预界说变量，如@hour， @daily等。更机动的方法是使用近来添加的时间表，它支持使用Python界说自界说时间表。
下面是怎样使用schedule_interval参数的示例。以下DAG将每天实行。

1. @dag(
2.     start_date=datetime(2023,1,1),
3.     schedule_interval = '@daily',
4.     catchup =False
5. )
6. def my_dag():
7.     pass

复制代码

关于调治，必要相识两个非常告急的概念：回填(backfill)和追赶(catchup)。
一旦我们界说了DAG，我们就设置了开始日期和操持隔断。如果catchup=True，则Airflow 将为从开始日期到当前日期的全部操持隔断创建DAG运行。如果catchup=False，气流将只从当前日期调治运行。
回填扩展了这个想法，使我们可以大概在CLI中创建已往的运行，而不管catchup参数的值：

1. $ airflow backfill  -s <START_DATE> -e <END_DATE> <DAG_NAME>

复制代码

毗连

Airflow 提供了一种简朴的方法来设置与外部体系或服务的毗连。可以使用UI、作为环境变量或通过设置文件创建毗连。它们通常必要URL、身份验证信息和唯一id。钩子(Hooks )是一种API，它抽象了与这些外部体系的通讯。比方，我们可以通过如下的UI界说一个PostgreSQL毗连：

然后使用PostgresHook来创建毗连并实行我们的查询：

1. pg_hook = PostgresHook(postgres_conn_id='postgres_custom')
3. conn = pg_hook.get_conn()
4. cursor = conn.cursor()
5. cursor.execute('create table _mytable (ModelID int, ModelName varchar(255)')
6. cursor.close()
7. conn.close()

复制代码

高级概念

为了使本教程尽大概完备，我必要提到一些更高级的概念。我不会具体先容每一个，但我猛烈发起你看看他们，如果你想深入把握Airflow 。

• 分支：分支允许你将使命分别为很多差别的使命，如：支持条件处置处罚差别使命的工作流。最常见的方法是BranchPythonOperator。
  
• 使命组：使命组可以在单个组中构造多个使命。它是简化图形视图和重复模式的好工具。
  
• 动态包：包和使命也可以以动态的方式构造。从Airflow 2.3开始，可以在运行时创建包和使命，这对于并行和依靠输入的使命来说是抱负的。气流也支持Jinja模板，而且是对动态包非常有用的增补。
  
• 单元测试和日记纪录：气流具有运行单元测试和纪录信息的专勤劳能.
  

Airflow最佳实践

在我们看到现实使用的示例之前，让我们讨论一下大多数从业者使用的一些最佳实践。

• 幂等性：dag和使命应该是幂等的。使用类似的输入重新实行类似的DAG运行应该始终具有与实行一次类似的效果。
  
• 原子性：使命应该是原子性的。每个使命应该负责一个使用，而且独立于其他使命
  
• 增量过滤：每个DAG运行应该只处置处罚一批支持增量提取和加载的数据。如许，大概出现的故障就不会影响整个数据集。
  
• 顶级代码：如果不是用于创建使用符或标志，则应克制使用顶级代码，由于它会影响性能和加载时间。全部代码都应该在使命内部，包罗导入包、数据库访问和繁重的盘算。
  
• 复杂性：dag应尽大概保持简朴，由于高复杂性大概会影响性能或调治。