编辑“︁Airflow与MySQL集成”︁

= Airflow与MySQL集成 =

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

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

== 前置条件 ==  
在开始之前，请确保：  
* 已安装并配置好 Airflow（建议版本 ≥ 2.0）。  
* 已安装 MySQL 服务器，并具备访问权限。  
* 已安装 Python 的 MySQL 客户端库（如 `mysql-connector-python` 或 `PyMySQL`）。  

== 安装依赖 ==  
在 Airflow 环境中安装 MySQL 连接器：  
<syntaxhighlight lang="bash">  
pip install apache-airflow-providers-mysql  
</syntaxhighlight>  

== 配置 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 示例：  

<syntaxhighlight lang="python">  
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');",  
    )  
</syntaxhighlight>  

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

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

<syntaxhighlight lang="python">  
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,  
)  
</syntaxhighlight>  

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

<syntaxhighlight lang="python">  
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,  
)  
</syntaxhighlight>  

== 常见问题与解决方案 ==  
=== 1. 连接失败 ===  
* 检查 MySQL 服务器是否运行。  
* 验证 Airflow 连接配置中的主机、端口和凭据。  

=== 2. 权限不足 ===  
* 确保 MySQL 用户具有执行 SQL 语句的权限。  

=== 3. 性能优化 ===  
* 对于大数据量操作，使用分批次处理或索引优化。  

== 总结 ==  
通过 Airflow 与 MySQL 的集成，可以实现数据管道的自动化管理。本教程介绍了基础配置、操作符使用、Python 交互以及实际案例，帮助用户快速上手。  

== 延伸阅读 ==  
* [[Airflow 官方文档]]  
* [[MySQL 官方文档]]

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow数据集成]]