Airflow与MySQL集成[编辑 | 编辑源代码]

介绍[编辑 | 编辑源代码]

Apache Airflow 是一个开源的工作流自动化工具，用于编排、调度和监控复杂的数据管道。MySQL 则是一个广泛使用的关系型数据库管理系统。将 Airflow 与 MySQL 集成，可以实现数据的自动化提取、转换和加载（ETL），从而构建高效的数据处理流程。

本教程将详细介绍如何在 Airflow 中连接 MySQL 数据库，执行 SQL 查询，并通过 DAG（有向无环图）调度任务。内容涵盖基础配置、常用操作符、实际案例以及常见问题解决方案。

前置条件[编辑 | 编辑源代码]

在开始之前，请确保：

已安装并配置好 Airflow（建议版本 ≥ 2.0）。
已安装 MySQL 服务器，并具备访问权限。
已安装 Python 的 MySQL 客户端库（如 `mysql-connector-python` 或 `PyMySQL`）。

安装依赖[编辑 | 编辑源代码]

在 Airflow 环境中安装 MySQL 连接器：

  
pip install apache-airflow-providers-mysql

配置 Airflow 连接[编辑 | 编辑源代码]

在 Airflow 的 Web UI 中配置 MySQL 连接： 1. 导航到 Admin → Connections。 2. 点击 "Add a new record"。 3. 填写以下信息：

  * Connection ID: `mysql_default`（或其他自定义名称）  
  * Connection Type: `MySQL`  
  * Host: MySQL 服务器地址（如 `localhost`）  
  * Schema: 默认数据库名称  
  * Login: MySQL 用户名  
  * Password: 密码  
  * Port: 3306（默认端口）

使用 MySQLOperator[编辑 | 编辑源代码]

Airflow 提供了 `MySqlOperator` 来执行 SQL 查询。以下是一个简单的 DAG 示例：

  
from datetime import datetime  
from airflow import DAG  
from airflow.providers.mysql.operators.mysql import MySqlOperator  

default_args = {  
    'owner': 'airflow',  
    'start_date': datetime(2023, 1, 1),  
}  

with DAG(  
    'mysql_example_dag',  
    default_args=default_args,  
    schedule_interval='@daily',  
) as dag:  

    create_table = MySqlOperator(  
        task_id='create_table',  
        mysql_conn_id='mysql_default',  
        sql="""  
        CREATE TABLE IF NOT EXISTS airflow_users (  
            id INT AUTO_INCREMENT PRIMARY KEY,  
            username VARCHAR(50) NOT NULL,  
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP  
        );  
        """,  
    )  

    insert_data = MySqlOperator(  
        task_id='insert_data',  
        mysql_conn_id='mysql_default',  
        sql="INSERT INTO airflow_users (username) VALUES ('airflow_user');",  
    )

代码说明[编辑 | 编辑源代码]

`mysql_conn_id` 指定在 Airflow 中配置的连接名称。
`sql` 参数接受要执行的 SQL 语句（支持多行）。

使用 PythonOperator 与 MySQL 交互[编辑 | 编辑源代码]

如果需要更灵活的操作，可以通过 `PythonOperator` 结合 Python 库（如 `PyMySQL`）与 MySQL 交互：

  
from airflow.operators.python import PythonOperator  
import pymysql  

def query_mysql():  
    connection = pymysql.connect(  
        host='localhost',  
        user='airflow',  
        password='airflow',  
        database='airflow_db'  
    )  
    cursor = connection.cursor()  
    cursor.execute("SELECT * FROM airflow_users;")  
    results = cursor.fetchall()  
    for row in results:  
        print(row)  
    connection.close()  

query_task = PythonOperator(  
    task_id='query_mysql',  
    python_callable=query_mysql,  
)

实际案例：数据备份与同步[编辑 | 编辑源代码]

以下是一个实际场景：每天将 MySQL 数据备份到 CSV 文件，并通过 Airflow 调度：

  
import pandas as pd  

def backup_mysql_data():  
    connection = pymysql.connect(  
        host='localhost',  
        user='airflow',  
        password='airflow',  
        database='airflow_db'  
    )  
    df = pd.read_sql("SELECT * FROM airflow_users;", connection)  
    df.to_csv('/tmp/airflow_users_backup.csv', index=False)  
    connection.close()  

backup_task = PythonOperator(  
    task_id='backup_data',  
    python_callable=backup_mysql_data,  
)