编辑“︁Airflow与数据库集成”︁（章节）

== 基础集成方法 ==

=== 使用数据库Operator ===

Airflow提供了多种数据库专用的Operator。以下是一个使用PostgresOperator执行SQL查询的示例：

<syntaxhighlight lang="python">
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime

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

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

    create_table = PostgresOperator(
        task_id='create_table',
        postgres_conn_id='postgres_conn',
        sql="""
        CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100),
            email VARCHAR(100)
        );
        """
    )

    insert_data = PostgresOperator(
        task_id='insert_data',
        postgres_conn_id='postgres_conn',
        sql="""
        INSERT INTO users (name, email)
        VALUES ('John Doe', 'john@example.com');
        """
    )

    create_table >> insert_data
</syntaxhighlight>

'''代码解释：'''
1. 创建了一个名为`postgres_example`的DAG
2. 定义了两个任务：`create_table`和`insert_data`
3. 使用`PostgresOperator`执行SQL语句
4. `postgres_conn_id`参数引用在Airflow中配置的数据库连接

=== 数据库连接配置 ===

在Airflow中使用数据库前，需要在Web UI或通过环境变量配置数据库连接：

1. 进入Airflow Web UI → Admin → Connections
2. 点击"Add a new record"
3. 填写连接信息：
   * Conn Id: `postgres_conn`
   * Conn Type: `Postgres`
   * Host: 数据库服务器地址
   * Schema: 数据库名称
   * Login: 用户名
   * Password: 密码
   * Port: 5432（PostgreSQL默认端口）