Apache Airflow 集成 Apache Pinot 实战：使用 SQLExecuteQueryOperator 执行实时 OLAP 查询

Apache Airflow 集成 Apache Pinot 实战：使用 SQLExecuteQueryOperator 执行实时 OLAP 查询

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

Apache Airflow 官方并没有为 Apache Pinot 提供专门的 Operator，而是统一推荐使用通用 SQL 执行算子SQLExecuteQueryOperator来对 Pinot Broker 发起标准 SQL 查询。本文以仓库中的 operators.rst 为核心，完整讲解连接配置、示例 DAG、底层 Hook 调用链与参数优先级，帮助你快速在 Airflow 中构建针对 Pinot 的查询任务，并掌握可复用的最佳实践。

为什么 Pinot 没有专属 Operator

Apache Pinot 是一个面向列的、分布式的开源 OLAP 数据存储（由 Java 编写），专为低延迟分析场景设计，适合在不可变数据上进行聚合类快速分析，也支持实时数据摄入。Pinot 对外暴露的是标准 SQL 查询接口，因此 Airflow 社区将其接入方式统一收敛到通用 SQL 体系：

• 使用SQLExecuteQueryOperator（位于airflow.providers.common.sql.operators.sql）执行查询；
• 底层由 Pinot Provider 提供的PinotDbApiHook负责建立与 Pinot Broker 的连接。

这一点在文档中有两条明确的note说明：

1. Apache Pinot 没有专属 Operator，请直接使用SQLExecuteQueryOperator；
2. 必须先安装对应的 Provider 包（apache-airflow-providers-apache-pinot），才能启用 Apache Pinot 支持。

从源码角度看，SQLExecuteQueryOperator继承自BaseSQLOperator，在execute阶段通过get_db_hook()获取对应的 DB Hook，再调用hook.run(...)执行 SQL（见 sql.py）。当conn_id指向一个pinot类型的连接时，框架会自动解析到PinotDbApiHook，这就是"无专属 Operator 却能开箱即用"的实现原理。

前置条件：安装 Provider 包

在已有的 Airflow 环境中执行：

pip install apache-airflow-providers-apache-pinot

根据 index.rst 中的要求，该 Provider 的最低依赖如下：

其中pinotdb是 Pinot 官方的 Python DB-API 驱动，PinotDbApiHook正是通过它连接 Broker。

配置 Pinot 连接（Connection）

使用SQLExecuteQueryOperator时，通过conn_id参数指向一个名为pinot类型的 Airflow Connection。连接元数据的结构如下：

注意事项：

• 文档表格中默认端口写为 8000，但实际生产环境常见的 Broker 端口是 8099 或 9000 等，请以你实际部署的 Pinot Broker 端口为准；
• Schema字段在查询场景下不被使用，但PinotDbApiHook.get_conn()会读取extra_dejson中的schema键作为请求协议（scheme），默认http；
• Extra中的endpoint用于指定 Broker 的 SQL 查询端点，默认值为/query/sql。

更深入的实现细节可以参考 pinot.py 中PinotDbApiHook.get_conn()的源码：它通过pinotdb.connect(host=..., port=..., username=conn.login, password=conn.password, path=..., scheme=...)建立连接。也就是说，Airflow Connection 的Login与Password字段会被透传给 Pinot 做身份认证，Extra中支持的键包括：

• endpoint：SQL 查询端点路径，默认query/sql；
• schema：连接协议，默认http（对应conn_type）。

此外，PinotDbApiHook.get_uri()会拼出形如http://localhost:9000/query/sql的 URI（conn_type://[login:password@]host:port/endpoint），可用于日志展示或诊断。

使用 SQLExecuteQueryOperator 编写查询 DAG

文档通过exampleinclude指令嵌入了系统测试示例 example_pinot.py，下面是其核心内容（对应[START howto_operator_pinot]与[END howto_operator_pinot]之间的代码）：

from __future__ import annotations import datetime from textwrap import dedent from airflow import DAG from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator DAG_ID = "example_pinot" with DAG( dag_id=DAG_ID, start_date=datetime.datetime(2025, 1, 1), default_args={"conn_id": "my_pinot_conn"}, schedule="@once", catchup=False, ) as dag: # Task: Simple query to test connection and query engine select_1_task = SQLExecuteQueryOperator( task_id="select_1", sql="SELECT 1", ) # Task: Count total records in airlineStats (sample table) count_airline_stats = SQLExecuteQueryOperator( task_id="count_airline_stats", sql="SELECT COUNT(*) FROM airlineStats", ) # Task: Group by Carrier and count flights group_by_carrier = SQLExecuteQueryOperator( task_id="group_by_carrier", sql=dedent(""" SELECT Carrier, COUNT(*) AS flight_count FROM airlineStats GROUP BY Carrier ORDER BY flight_count DESC LIMIT 5 """).strip(), ) select_1_task >> count_airline_stats >> group_by_carrier

代码要点拆解

1. conn_id 的两种指定方式：示例在 DAG 的default_args中统一设置了"conn_id": "my_pinot_conn"，因此每个SQLExecuteQueryOperator无需重复传参；你也可以在单个 Task 上通过conn_id=显式覆盖。
2. sql参数支持多行 SQL：SQLExecuteQueryOperator的template_fields = ("sql", "parameters", ...)，说明sql是支持模板渲染的字段。多行语句可以用textwrap.dedent或三引号书写，保持可读性。
3. 任务编排：示例用位运算>>将三个查询串成线性链：select_1（连通性测试）→count_airline_stats（全表计数）→group_by_carrier（分组聚合）。其中SELECT 1常被用来快速验证连接与查询引擎是否正常。

算子关键参数速查

结合SQLExecuteQueryOperator的源码定义（sql.py），除conn_id外，最常用的参数有：

注意：PinotDbApiHook声明了supports_autocommit = False，且其set_autocommit、insert_rows均直接抛出NotImplementedError——这印证了 Pinot 集成定位是只读分析查询，不适合通过该 Hook 写入数据。

参数优先级：算子入参 > 连接元数据

文档末尾特别强调：

Parameters provided directly viaSQLExecuteQueryOperator()take precedence over those specified in the Airflow connection metadata.

即直接在SQLExecuteQueryOperator()中传入的参数，优先级高于 Airflow 连接元数据中的配置。这一点是 Airflow 通用 SQL 体系的统一行为：Connection 提供的是"默认连接信息"，而每个 Task 上的显式参数可以在不修改 Connection 的前提下覆盖默认值。例如可以在 Connection 中配置默认的 Broker 地址，而在特定 Task 上通过参数临时指向其他 Broker 实例。

底层链路：从 Operator 到 Pinot Broker

一次查询任务的完整调用链可以概括为：

SQLExecuteQueryOperator.execute() │ 1. get_db_hook() 解析 conn_id（pinot 类型 → PinotDbApiHook） ▼ PinotDbApiHook.run(sql, ...) # 继承自 DbApiHook │ 2. get_conn() 调用 pinotdb.connect(...) ▼ pinotdb.Connection → Cursor.execute(sql) │ 3. 向 http://<host>:<port>/query/sql 发起标准 SQL 请求 ▼ Pinot Broker（标准 SQL 查询端点，即 PQL 端点弃用后的替代方案）

几个值得注意的实现事实：

• PinotDbApiHook明确注释：使用标准 SQL 端点，因为 PQL 端点即将被弃用（对应官方查询文档）；conn_type = "pinot"、hook_name = "Pinot Broker"，这些注册信息同时出现在 get_provider_info.py 与 provider.yaml 中；
• 连接信息中的login/password会作为username/password传给pinotdb.connect，可用于带认证的 Broker；
• SQLExecuteQueryOperator.execute()在do_xcom_push或requires_result_fetch为真时才传入handler抓取结果，否则查询结果不会显式拉取，这一点对超大结果集的查询可以起到节省内存的作用。

延伸：用 Hook 完成离线数据管理

虽然本文主题是查询算子，但 Pinot Provider 还提供PinotAdminHook，用于调用pinot-admin.sh脚本完成离线数据摄入（AddSchema、AddTable、CreateSegment、UploadSegment 四个子命令），可作为构建"离线灌数 + 在线查询"完整链路时的补充手段，详见 hooks.rst。PinotAdminHook的关键行为包括：

• 在 4.0.0 版本起cmd_path被硬编码为pinot-admin.sh，必须确保该脚本在 PATH 中，传入其他值会直接抛出RuntimeError；
• 由于早期 Pinot 的pinot-admin.sh无论成败都返回退出码 0，可通过pinot_admin_system_exit标志（或连接Extra中的同名键）切换为"按输出内容判断"：当输出中包含Error或Exception时视为失败并抛出AirflowException。

结语

Apache Airflow 与 Apache Pinot 的集成遵循"通用 SQL 优先"的设计哲学：没有专属 Operator，不代表能力缺失。借助SQLExecuteQueryOperator+PinotDbApiHook，你可以在一个统一、可模板化、支持 XCom 传递结果的框架下，对 Pinot 执行标准 SQL 聚合查询。实际操作时请记住三件事：安装apache-airflow-providers-apache-pinot并确保pinotdb依赖可用、按 Broker 实际情况配置pinot类型 Connection（Host/Port/Extra 端点）、利用算子显式参数覆盖连接默认值。系统测试示例 example_pinot.py 与 example_pinot_dag.py 是开箱即用的参考实现，可直接作为新 DAG 的起点。

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow