使用CDK部署基于Fargate的高可用、易扩展的Airflow集群

前言

Apache Airflow(以下简称为Airflow) 是一项由 Airbnb 在 2014 年推出的开源项目，其目的是为了管理日益复杂的数据管理工具、脚本和分析工具，提供一个构建批处理工作流的方案。Airflow是一种可扩展的分布式工作流调度系统，允许将工作流建模为有向无环图（DAGs），通过这种方式简化数据管道中各个处理步骤的创建、编排和监控。Airflow被广泛应于于机器学习、数据分析/处理以及各类需要流程化的场景。

在2020年初的时候，费老师写了一篇Blog 来剖析Airflow架构，并给出了在亚马逊云科技上部署Airflow的建议。也正因为Airflow在机器学习，数据分析等领域用户众多，为了满足这部分用户的需要，亚马逊云科技在2020年11底推出了托管的Apache Airflow 服务 Managed Workflows for Apache airflow (MWAA)。

目前MWAA已经在亚马逊云科技全球大部分区域可用，如果要在尚未发布的区域使用Airflow，或者有定制化部署Airflow的需求，可以使用自建部署的方式。然而，部署Airflow并不是一件简单的事情，如何做好高可用，易扩展，如何保障安全，以及减少运维难度等有很多细节。本文以在亚马逊云科技中国区部署自建Airflow为例，提出了一个参考架构，并提供了一套CDK的参考实现，让使用AMAZON WEB SERVICES CDK的用户通过数行代码就可以将整套方案部署到云上。

架构考量

Airflow 体系架构概述

如下图所示，Airflow 主要组成部分为：

Webserver：为用户提供界面用来管理，调度编排任务；
Scheduler：处理触发预定的工作流，并将任务提交给执行器运行;
Worker：负责处理运行任务；
DAG Directory: 用来存放用户定义的DAG文件；
Metadata Database: 用来存储集群的元数据；

用户将DAG文件上传到 DAG Directory，登陆到Webserver，可以管理，调度编排任务。用户的操作被写入Metadata Database，Metadata Database记录着DAG中的任务以及后台对应的状态，Scheduler 从Metadata Database读取数据，检查每个任务的状态，并将任务调度给Worker，Worker执行任务，更新Metadata Database中的状态信息。

Airflow On AMAZON WEB SERVICES 架构概述

Airflow在亚马逊云科技上进行部署时：

Webserver/Scheduler/Worker三个组件，可以部署到Fargate上，充分利用无服务器架构的优势，减少维护工作量；

DAT Directory使用S3，用户将DAG文件上传到S3，webserver/scheduler 将DAG文件同步到集群；

Metadata Database使用托管的数据库服务RDS，提高可用率，减少维护工作量。

整体架构图如下所示:

如下为详细的设计考量：

网络层

使用三层经典架构。

Airflow Fargate集群位于 Private subnet，Airflow 访问Public ECR Repository通过 NAT Gateway–> Internet Gateway 访问，访问ECS Control Plane， CloudWatch， SecretsManager 则通过 Interface Endpoint走内网，访问S3通过Gateway Endpoint走内网；

ElasticCache和RDS位于Private Subnet，通过Security Group的限制，只对Fargate集群开放。

存储层

数据库选择使用RDS for Postressql，保证服务高可用，降低运维工作量；

内存数据库选择使用Elastic for Redis，作为缓存；

DAG Directory使用S3，利用S3的持久性和便利性，用户将DAG文件上传到S3，Fargate定时同步DAG文件。

计算层

计算层选择Amazon ECS on Fargate，您无需预置和管理服务器，而且可以为Airflow的Webserve/Scheduler/Worker指定独立的资源，并通过设计隔离应用程序来提高安全性。Fargate 分配了适当的计算量，从而无需选择实例和扩展集群容量。

服务层

Airflow 的Scheduler 和 Worker的运行依赖于Webserver，选择使用 AMAZON WEB SERVICES CloudMap 来做服务注册和服务发现，维护不断变化的资源的更新位置，提高应用程序的可用性。

Airflow On AMAZON WEB SERVICES 部署概述

基于如上方案，将Airflow部署到云上，大致需要如下步骤:

创建 S3 Bucket，用于存放用户上传的DAG文件；
创建VPC，并确保：
- VPC 跨最少两个可用区；
- VPC 打开 enableDNSHostNames 功能；
- VPC 打开 enableDNSSupport功能；
创建Security Group，具体包括:
- VPC Endpoint Security Group
- Airflow ECS Service Security Group
- Database Security Group
- Redis Security Group
- ELB Security Group
创建 VPC Endpoint，具体包括：
- S3 Gateway Endpoint
- ECS/ECR/CloudWatch/SecretsManager Interface Endpoint；
创建RDS 数据库
在Secrets Manager中生成Database Credentials，并应用于RDS数据库；
创建Redis集群；
创建CloudWatch LogGroup；
创建ECS 集群；
创建并配置Cloud Map；
创建Load Balancer, TargetGroup, Airflow Webserver ECS Task Definition, Airflow Webserver ECS Service;
创建Airflow Scheduler ECS Task Definition, Airflow Scheduler ECS Service;
创建Airflow Worker ECS Task Definition, Airflow Worker ECS Service;

可以看出，在云上部署高可用，易扩展的Aiflow集群，并不是一简单的事情。

因此，这里利用AMAZON WEB SERVICES CDK，将上述工作封装成了一L2的Construct Library，用户使用CDK只需要简单的数行代码即可在云上完成上述工作。

基于AMAZON WEB SERVICES CDK部署Airflow

AMAZON WEB SERVICES CDK 开发环境搭建

开发AMAZON WEB SERVICES CDK需要先安装AMAZON WEB SERVICES CDK CLI，利用 AMAZON WEB SERVICES CDK CLI可以生成对应的CDK 的Project。

AMAZON WEB SERVICES CDK CLI的安装依赖于Node.js, 所以在您的开发环境需要先安装node.js。node.js 的安装可参看官方教程: https://nodejs.org/en/download/package-manager/。

安装好 node.js 之后，可以直接使用如下命令安装 AMAZON WEB SERVICES CDK CLI：

npm install -g aws-cdk #安装cdk cli
cdk –version #查看版本

使用AMAZON WEB SERVICES CDK 创建Apache Airflow

安装 CDK CLI后，在您的环境中通过命令 aws configure 命令配置好对应的权限之后，就可以使用AMAZON WEB SERVICES CDK来进行开发了。

AMAZON WEB SERVICES CDK 支持 JavaScript、TypeScript、Python、Java、C# 和 Go 等开发语言，这里以 TypeScript为例。

在命令行中运行如下命令，创建CDK Project:

mkdir apache-airflow-cdk && cd apache-airflow-cdk
cdk init sample-app –language=typescript

在IDE（这里以VSCode为例）打开文件夹，如下图所示：

在 lib文件夹下的apache-airflow-cdk-stack.ts定义了CloudFormation对应的结构，可以看出，默认的Sample中创建了一个SQS的Queue，和一个SNS的Topic。

删除掉Sample代码，修改代码为如下，创建Airflow 集群:

import * as cdk from '@aws-cdk/core';
import * as airflow from 'cdk-serverless-airflow';

export class ApacheAirflowCdkStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
    super(scope, id, props);
    
    new airflow.Airflow(this, 'Airflow');
    
  }
}