运行 SQL 工作负载

使用 Airflow 构建数据流水线

Volker Janz

Senior Developer Advocate at Astronomer

为何在 Airflow 中用 SQL？

SQL 工作负载是最常见的 Airflow 用例
让数据库承担重计算
Airflow 负责编排何时与何处
数据库负责 如何执行

Airflow 中的 SQL 编排

SQLExecuteQueryOperator

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


SQLExecuteQueryOperator(
    task_id="load_sales",
    conn_id="duckdb_analytics",
    sql="INSERT INTO sales SELECT * FROM staging",
)

适用于具有兼容 Airflow provider 的任意数据库
PostgreSQL、Snowflake、BigQuery、DuckDB 等

连接（Connections）

将凭据保存在代码之外
每个都有 ID、类型、主机、端口、登录名
可通过多种方式创建：UI、CLI、API 或环境变量
在算子中通过 conn_id 引用

Airflow Connections 界面

构建 SQL 流水线

from airflow.sdk import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

@dag(schedule="@daily")
def sales_pipeline():

    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="""
            INSERT INTO daily_summary (order_date, total_orders, total_revenue)
            SELECT order_date, COUNT(*), SUM(revenue)
            FROM raw_orders GROUP BY order_date
        """,
    )

外部 SQL 文件

@dag(
    schedule="@daily",
    template_searchpath="/path/to/include/sql",
)
def sales_pipeline():
    aggregate = SQLExecuteQueryOperator(
        task_id="aggregate_daily_sales",
        conn_id="duckdb_analytics",
        sql="aggregate_sales.sql",
    )

在 @dag 上设置 template_searchpath
在 sql 中引用文件名
将脚本与业务逻辑放在 dags/ 文件夹之外 💡

项目文件结构

SQL 文件中的 Jinja 模板

DELETE FROM daily_summary WHERE order_date = '{{ ds }}';

INSERT INTO daily_summary (order_date, total_orders, total_revenue)
SELECT order_date, COUNT(*), SUM(revenue)
FROM raw_orders
WHERE order_date = '{{ ds }}'
GROUP BY order_date;

{{ ds }} 渲染为逻辑日期（YYYY-MM-DD）
先 DELETE 后 INSERT：第 2 章的幂等模式
重新运行同一日期会得到相同结果

params 与 parameters 的区别

params（Jinja 渲染）

SQLExecuteQueryOperator(
    sql="""SELECT * FROM orders
           WHERE product = '{{ params.product }}'""",
    params={"product": user_input},
)

值被插入到 SQL 字符串中
易受SQL 注入影响

parameters（数据库级绑定）

SQLExecuteQueryOperator(
    sql="SELECT * FROM orders
         WHERE product = $product",
    parameters={"product": user_input},
)

值传递给数据库驱动
防注入：由驱动处理转义

用 Astro 从本地到生产

Astronomer's Astro：托管的 Airflow 平台

Astro CLI：一条命令启本地 Airflow

本地开发，顺畅部署到生产

Astro CLI 与平台

让我们一起练习吧！

使用 Airflow 构建数据流水线