编辑“︁Airflow与RDS交互”︁

= Airflow与RDS交互 =

== 介绍 ==  
'''Amazon Relational Database Service (RDS)''' 是一种托管式关系型数据库服务，支持多种数据库引擎（如MySQL、PostgreSQL、Oracle等）。Apache Airflow 作为工作流编排工具，可通过任务（Task）与RDS交互，实现数据提取、转换、加载（ETL）、定时备份等自动化操作。本指南将介绍如何在Airflow中配置RDS连接、执行SQL操作，并通过实际案例展示典型应用场景。

== 核心概念 ==  
=== 1. Airflow连接机制 ===  
Airflow通过'''Connection'''对象管理外部系统（如RDS）的认证信息。连接参数包括：  
* '''Host''': RDS实例端点  
* '''Port''': 数据库端口（如MySQL默认3306）  
* '''Login''' & '''Password''': 数据库凭据  
* '''Schema''': 默认数据库名称  

=== 2. 数据库Hook ===  
Airflow提供'''BaseHook'''及其子类（如'''PostgresHook''', '''MySqlHook'''）封装数据库操作，简化连接管理和SQL执行。

== 配置RDS连接 ==  
=== 通过Airflow UI ===  
1. 登录Airflow Web UI → '''Admin''' → '''Connections'''  
2. 点击"+"新建连接，填写RDS信息：  
   * Conn Id: `rds_mysql_conn`  
   * Conn Type: `MySQL`  
   * Host: `your-rds-endpoint.rds.amazonaws.com`  
   * Schema: `target_database`  
   * Login: `admin`  
   * Password: `secure_password`  
   * Port: `3306`  

=== 通过环境变量 ===  
在`airflow.cfg`或部署时设置：  
<syntaxhighlight lang="bash">  
export AIRFLOW_CONN_RDS_MYSQL_CONN="mysql://admin:secure_password@your-rds-endpoint.rds.amazonaws.com:3306/target_database"  
</syntaxhighlight>  

== 代码示例 ==  
=== 示例1：使用MySqlHook查询数据 ===  
<syntaxhighlight lang="python">  
from airflow import DAG  
from airflow.providers.mysql.hooks.mysql import MySqlHook  
from airflow.operators.python import PythonOperator  
from datetime import datetime  

def fetch_data():  
    hook = MySqlHook(mysql_conn_id='rds_mysql_conn')  
    sql = "SELECT * FROM sales WHERE date >= '2023-01-01';"  
    df = hook.get_pandas_df(sql)  
    print(f"Fetched {len(df)} rows")  

with DAG('rds_interaction_dag', start_date=datetime(2023,1,1)) as dag:  
    query_task = PythonOperator(  
        task_id='fetch_sales_data',  
        python_callable=fetch_data  
    )  
</syntaxhighlight>  

'''输出说明'''：  
任务会打印查询到的行数，数据以Pandas DataFrame形式返回。

=== 示例2：使用PostgresHook执行事务 ===  
<syntaxhighlight lang="python">  
from airflow.providers.postgres.hooks.postgres import PostgresHook  

def update_inventory():  
    hook = PostgresHook(postgres_conn_id='rds_postgres_conn')  
    conn = hook.get_conn()  
    cursor = conn.cursor()  
    try:  
        cursor.execute("UPDATE inventory SET stock = stock - 1 WHERE product_id = 101;")  
        cursor.execute("INSERT INTO orders (product_id, quantity) VALUES (101, 1);")  
        conn.commit()  
    except Exception as e:  
        conn.rollback()  
        raise e  

# 在DAG中调用此函数  
</syntaxhighlight>  

'''关键点'''：  
* 显式管理事务（commit/rollback）  
* 适用于需要原子性的操作  

== 实际案例 ==  
=== 场景：每日销售报表生成 ===  
1. **流程设计**：  
<mermaid>  
graph TD  
    A[启动DAG] --> B[从RDS提取当日订单]  
    B --> C[计算销售汇总]  
    C --> D[保存结果到S3]  
    D --> E[发送邮件通知]  
</mermaid>  

2. **实现代码片段**：  
<syntaxhighlight lang="python">  
def generate_report():  
    mysql_hook = MySqlHook('rds_mysql_conn')  
    summary_sql = """  
        SELECT product_id, SUM(amount) as total_sales  
        FROM orders  
        WHERE order_date = CURDATE()  
        GROUP BY product_id;  
    """  
    summary = mysql_hook.get_records(summary_sql)  
    # 后续处理...  
</syntaxhighlight>  

== 性能优化建议 ==  
* 使用'''XCom'''传递小规模数据，避免大结果集直接传输  
* 对高频查询启用RDS '''读写分离'''  
* 通过'''Batch Operator'''批量处理数据（如`MySqlOperator`的`sql`参数支持列表）  

== 常见问题 ==  
'''Q: 如何安全存储RDS密码？'''  
A: 使用Airflow的'''Vault'''或AWS '''Secrets Manager'''后端，避免明文存储。  

'''Q: 连接超时如何解决？'''  
A: 检查：  
1. RDS安全组是否允许Airflow Worker IP  
2. 连接池配置（如`extra`参数添加`{"pool_size": 10}`）  

== 数学表达（可选） ==  
对于需要计算吞吐量的场景：  
<math>  
\text{Throughput} = \frac{\text{Queries Processed}}{\text{Time Period}}  
</math>  

== 总结 ==  
通过Airflow与RDS集成，可实现可靠的数据库自动化运维。关键步骤包括：  
# 正确配置连接  
# 选择合适的Hook/Operator  
# 实现错误处理机制  
# 监控任务执行（如通过Airflow日志）  

下一步可探索：  
* 动态生成SQL参数（使用Jinja模板）  
* 与AWS其他服务（如Lambda、Redshift）联动

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