使用SAE部署Airflow

1.1 创建MySQL实例

登录云数据库 RDS 控制台，在左侧导航栏选择实例列表，单击创建实例，然后配置以下参数。

1. 选择创建方式，支持标准创建和快捷创建，本文以标准创建为例进行演示。

2. 选择付费方式，选择与SAE侧部署Airflow应用相同的地域，选择引擎为MySQL并选择符合要求的版本。

   

3. 选择VPC，使用与部署Airflow应用所在命名空间相同的VPC。

4. 选择主可用区及网络、备用可用区及网络。

5. 选择实例规格。

6. 单击确定。

1.2 修改参数

1. 在实例列表页面，单击已创建的实例ID，进入实例的基本信息页面。

2. 在左侧导航栏单击参数设置，在可修改参数页签，找到sql_mode参数。

3. 单击sql_mode参数运行参数值列的图标，删除NO_ZERO_DATE参数值，单击确定，然后单击提交参数。

   说明

   在严格模式下，MySQL数据库不允许0000-00-00作为有效时间。在某些情况下，会有类似“Invaliddefaultvaluefor‘end_date'”的报错，该报错表示某些Airflow表使用0000-00-00 00:00:00作为时间戳字段的默认值，在MySQL实例上禁用NO_ZERO_DATE参数值后，可避免此错误。

   

1.3 创建账号及数据库

本文通过控制台的方式为例进行演示说明。

CREATE DATABASE airflow_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
# 创建airflow_db数据库

CREATE USER 'airflow_user' IDENTIFIED BY 'airflow_pass';
# 创建登录数据库的账号

GRANT ALL PRIVILEGES ON airflow_db.* TO 'airflow_user';
# 授予airflow_user用户管理airflow_db数据库的权限

1.4 设置数据库白名单

1. 进入实例后，在左侧导航栏单击白名单与安全组。

2. 在名单设置页签，单击添加白名单分组。

3. 在弹出的添加白名单分组面板，设置分组名称和组内白名单，然后单击确定。

   说明

   • 如果需要添加多个网段，需要使用半角逗号（,）进行分隔。

   • 新白名单将于1分钟后生效。

   • 分组名称可自定义。

   • 组内白名单为SAE侧部署Airflow应用对应命名空间的VPC可交换机的网段。

     • 查看VPC网段：在SAE侧，进入部署Airflow应用的目标命名空间，单击专有网络VPC名称，然后在目标VPC的基本信息页签复制IPv4网段。

       

     • 查看交换机网段：在专有网络控制台，单击左侧导航栏的交换机，在交换机页面找到对应的交换机并单击实例ID，然后复制该交换机的IPv4网段。

       

1.5 获取数据库连接地址

进入实例后，在左侧导航栏单击数据库连接，然后复制该页面的内网地址和内网端口。

创建Airflow应用时，需要填写数据库连接地址。

2.1 创建Bucket

创建存储空间的方式有多种，本文以通过控制台创建为例进行演示。具体信息，请参见创建存储空间。

1. 登录对象存储OSS控制台，在左侧导航栏单击Bucket列表。

2. 单击创建Bucket，在弹出的创建Bucket面板配置以下参数，然后单击完成创建。

   1. 选择创建模式。

   2. 自定义Bucket名称，并选择目标地域。

      选择的地域与创建Airflow应用的地域一致。

   3. 选择所需资源组。

2.2 创建挂载目录

1. 在Bucket列表页面，单击上一步骤创建的Bucket名称。

2. 在目标Bucket的文件列表页面，单击新建目录。

3. 在弹出的新建目录页签，输入目录名，然后单击确定。

   说明

   目录命名规范：

   • 不允许使用表情符，请使用符合要求的 UTF-8 字符。

   • /用于分割路径，可快速创建子目录，但不要以/或\开头，不要出现连续的/。

   • 不允许出现名为..的子目录。

   • 总长度控制在1~254 个字符。

3.1 配置网络

1. 登录SAE控制台，在左侧导航栏单击应用中心，选择目标地域。

   部署Airflow应用的地域和数据库所在的地域需要一致。

2. 然后将鼠标悬停到Airflow区域，单击部署。

3. 在Airflow页面，选择命名空间、交换机和选择安全组。

   为了保证服务的高可用，建议至少选择两个交换机。需要将选择的交换机的网段都添加到数据库的白名单中。

   

3.2 配置MySQL数据库

在Airflow页面，单击数据库区块，在数据库配置面板配置以下信息，然后单击确定。

1. 数据库名称：填写已创建的数据库名称。

2. 数据库地址：填写已复制的内网连接地址。

3. 端口：填写数据库的端口，默认为3306，如果您自定义了数据库端口，请填写自定义的端口。

4. 用户名：填写用于登录数据库的账号名称。

5. 密码：填写用于登录数据库的账号密码。

3.3 配置文件存储

在Airflow页面，单击文件存储区块，在文件存储配置面板配置以下信息，然后单击确定。

1. 填写已创建的AccessKey ID和AccessKey Secret。

   建议您遵循阿里云安全最佳实践，使用RAM子用户的AccessKey，并设置RAM最小访问权限。具体操作，请参见创建AccessKey。

2. 挂载OSS。

   1. Bucket：在下拉列表中选择已创建的Bucket。

   2. 挂载路径：填写已创建的挂载目录，如test/。

      如果OSS挂载目录不存在，会触发异常。

   3. 容器路径：填写/opt/airflow/dags。

3.4 执行流水线部署

数据库和文件存储配置完成后，在Airflow页面单击部署。

• 执行流水线部署后，会创建sae-airflow-scheduler和sae-airflow-webserver两个应用，此两个应用可在应用列表页面查看和管理。

  

• 流水线部署过程中，会实时打印部署流水线的日志，包括Secret数据库连接、Deployment应用发布和Service公网访问发布过程，如果流水线发布失败，您可以借助发布日志排查问题。

• 流水线部署完成后，会在Service公网访问发布页签打印sae-airflow-webserver应用的CLB IP和端口，该地址可直接在浏览器访问。

  

4.1 创建Airflow用户名和密码

流水线发布成功后，需要创建登录Airflow Web界面的用户名和密码。

1. 在应用列表页面，进入sae-airflow-scheduler或sae-airflow-webserver应用的基础信息页面。

2. 在基础信息页面，单击实例列表页签，使用Websell登录应用实例，然后创建用户名和密码。

   airflow users create \
      --username airflow \
      # 自定义用户名，本示例中为airflow
      --password airflow \
      # 自定义密码，本示例中为airflow
      --role Admin \
      # 定义用户角色，Admin为最高权限，可管理所有DAG和用户
      --firstname airflow \
      --lastname airflow \
      --email spider***@superhero.org 
      # 定义用户邮箱，用于接收通知，如任务失败告警

4.2 登录Airflow Web界面

复制在流水线中生成的公网IP，在浏览器在进行访问，然后输入用户名和密码进行登录。

格式为http://<sae-airflow-webserver CLB IP>:8080

4.3 体验运行DAG任务

1. 登录OSS管理控制台，在已创建的Bucket中上传示例DAG文件，以便调度器识别并运行任务。

   上传文件的具体操作，请参见上传文件。

   示例文件：dag.py

   DAG示例文件内容

   from airflow import DAG
   from airflow.operators.bash_operator import BashOperator
   from airflow.operators.python_operator import PythonOperator
   from datetime import datetime, timedelta
   
   def my_python_function():
       print("Hello from Python!")
   
   
   def my_python_function1():
       print("Hello from airflow!")
   
   # 创建 DAG 对象
   with DAG(
       "example_dag",
       schedule_interval='*/1 * * * *',  # 每分钟执行一次
       start_date=datetime(2022, 1, 1),
       catchup=False,
   ) as dag:
       task1 = PythonOperator(
           task_id="extract_data",
           python_callable=my_python_function1,
           dag=dag,
       )
       task2 = PythonOperator(
           task_id="my_python_task",
           python_callable=my_python_function,
           dag=dag,
       )
       task1 >> task2  # 定义任务的依赖关系

2. 进入Airflow Web界面，在DAGs页面，单击目标DAG名称。

   

3. 在目标DAG详情页面，单击Graph页签。

   下图显示的是两个DAG对象，分别对应的是task1和task2。由示例文件可知，task2依赖于task1，即task1运行成功完成后才会执行task2。

   

4. 单击Graph页签中的print_date，运行DAG任务，查看是否能够运行成功。

   运行成功后，DAG对象会显示success。

   

5. 在DAG对象运行过程中，会在sae-airflow-scheduler应用中创建对应的实例，DAG对象运行成功后，对应的实例会自动删除。

   说明

   • Airflow Worker实例归属于airflow-scheduler组件，可以在sae-airflow-scheduler应用的实例列表页面查看。

   • 如果有Pod泄露，您可以直接在sae-airflow-scheduler应用的实例列表页面查看与删除。