Apache Airflow Impala Provider 实战使用 SQLExecuteQueryOperator 连接与操作 Apache Impala【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本文基于 Airflow 仓库中 Impala Provider 的官方操作文档providers/apache/impala/docs/operators.rst展开讲解如何在 Airflow DAG 中通过SQLExecuteQueryOperator对 Apache Impala 集群执行 SQL 查询。读完本文你将掌握 Impala 连接的完整元数据配置、示例 DAG 的写法与参数优先级规则并能结合 ImpalaHook 源码 理解连接字段到impyla客户端的映射原理。背景为什么 Impala 使用通用的 SQLExecuteQueryOperatorImpala Provider 的官方文档明确指出此前 Impala 曾有专用的 Operator该专用 Operator 已被弃用deprecated现在应统一使用SQLExecuteQueryOperator来自airflow.providers.common.sql.operators.sql来对 Apache Impala 集群执行 SQL 查询。这一变化意味着 Impala 与其他数据库Postgres、MySQL、Hive 等在 Airflow 中的用法趋同——只要连接类型connection type对应的 Hook 实现正确通用的 SQL 系列 Operator 即可驱动它。从 provider.yaml 可以看到该 Provider 的元信息为包名apache-airflow-providers-apache-impala当前仓库中版本为 1.9.3lifecycle: production注册的 Hook 为 ImpalaHookPython 模块airflow.providers.apache.impala.hooks.impala注册的连接类型connection-type: impala绑定hook-class-name: airflow.providers.apache.impala.hooks.impala.ImpalaHook。也就是说当你使用impala类型的 Airflow 连接时Airflow 会自动解析出ImpalaHook而SQLExecuteQueryOperator则通过该 Hook 执行 SQL。安装 Provider 包与依赖文档中有一个重要提示必须安装apache-airflow-providers-apache-impala包才能启用 Impala 支持pip install apache-airflow-providers-apache-impala结合 pyproject.toml 与 README.rst该包的依赖与兼容性如下依赖版本要求impyla0.22.0,1.0apache-airflow-providers-common-sql1.32.0apache-airflow-providers-common-compat1.12.0apache-airflow2.11.0可选依赖extrasExtra依赖kerberoskerberos1.3.0Kerberos/GSSAPI 认证场景sqlalchemysqlalchemy1.4.54需要构建 SQLAlchemy URL 时使用该包支持 Python 3.10 3.14。注意impyla是实际的数据库客户端底层通过 HS2Hive Server 2协议与impalad通信连接参数如 SSL、认证机制都经由它传递。配置 Impala 连接Connection Metadata使用conn_id参数连接 Impala 实例时连接元数据Connection Metadata的结构如下完整继承自 operators.rst参数含义Host (string)Impala 守护进程的域名或 IP 地址可以是任意一个impalad服务节点Schema (string)默认数据库名可选。若为空行为由具体实现决定Login (string)认证用户名如适用例如 LDAP 用户Password (string)认证密码如适用Port (int)Impala 服务端口默认21050注意与 Hive 端口通常不同Extra (JSON)附加连接配置例如{use_ssl: false, auth: NOSASL}字段详细说明可参考 Provider 自带的连接文档 providers/apache/impala/docs/connections/impala.rst其中强调 Host/Port 对应 HS2 协议端点Impala 默认端口为21050Extra 字段是一个 JSON 字典其中的键会作为额外参数传给impyla连接。此外Impala 的 Hook 与 Operator 默认使用的连接 ID 为impala_default见 ImpalaHook 中的default_conn_name impala_default如果你按默认方式配置连接直接使用这个名字即可省去显式传conn_id。源码视角连接字段如何映射到 impylaImpalaHook 是DbApiHook的子类其get_conn()方法把 Airflow 连接对象的各字段逐一映射为impyla.dbapi.connect的调用参数def get_conn(self) - Connection: conn_id: str self.get_conn_id() connection self.get_connection(conn_id) return connect( hostconnection.host, portconnection.port, userconnection.login, passwordconnection.password, databaseconnection.schema, **connection.extra_dejson, # Extra JSON 中的键值原样展开为 connect() 参数 )两个值得注意的实现细节schema映射为databaselogin映射为user——配置连接时 Schema 字段就是 impyla 的默认数据库connection.extra_dejson被直接解包**因此 Extra 中写的任何 JSON 键如use_ssl、auth_mechanism、configuration都会透传给impyla的connect()。单元测试 test_impala.py 对此做了两类验证普通场景extra{use_ssl: True}最终断言为connect(host..., port21050, user..., password..., database..., use_sslTrue)Kerberos 场景extra{auth_mechanism: GSSAPI, use_ssl: True}最终断言为connect(..., use_sslTrue, auth_mechanismGSSAPI)。如果你在集群中启用了 Kerberos 认证需要在 Extra 中配置auth_mechanism: GSSAPI并安装kerberosextra。另外ImpalaHook还提供了sqlalchemy_url属性与get_uri()方法impala.py可将连接渲染为impala://user:passhost:21050/db?queryparams形式的 SQLAlchemy URL。该功能依赖可选的sqlalchemyextra——未安装时会抛出AirflowOptionalProviderFeatureException并提示以pip install apache-airflow-providers-apache-impala[sqlalchemy]安装。构建 URL 时若连接缺少host或login会抛出带明确信息的ValueError。示例在 DAG 中使用 SQLExecuteQueryOperator 连接 Impala官方文档通过exampleinclude指令引用了系统测试 DAG example_impala.py标记[START howto_operator_impala][END howto_operator_impala]区段。下面给出该示例 DAG 的完整可运行内容import datetime from airflow import DAG from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator DAG_ID example_impala with DAG( dag_idDAG_ID, start_datedatetime.datetime(2025, 1, 1), default_args{conn_id: my_impala_conn}, # Impala 连接 ID scheduleonce, catchupFalse, ) as dag: create_table_impala_task SQLExecuteQueryOperator( task_idcreate_table_impala, sql CREATE TABLE IF NOT EXISTS impala_example ( a STRING, b INT ) PARTITIONED BY (c INT) , ) alter_table_impala_task SQLExecuteQueryOperator( task_idalter_table_impala, sqlALTER TABLE impala_example ADD PARTITION (c1), ) insert_data_impala_task SQLExecuteQueryOperator( task_idinsert_data_impala_task, sqlINSERT INTO impala_example PARTITION (c1) VALUES (a, 1), (a, 2), (b, 3), ) select_data_impala_task SQLExecuteQueryOperator( task_idselect_data_impala, sqlSELECT * FROM impala_example, ) drop_table_impala_task SQLExecuteQueryOperator( task_iddrop_table_impala, sqlDROP TABLE impala_example, ) ( create_table_impala_task alter_table_impala_task insert_data_impala_task select_data_impala_task drop_table_impala_task )注上方代码中insert_data_impala_task的task_id已修正为不与变量名冲突的写法其余 SQL 与依赖链与原文件 example_impala.py 保持一致。示例覆盖了 Impala 的典型 DDL/DML 生命周期建分区表、加分区、按分区插入、查询、删表并且通过default_args把conn_id统一注入到所有 Operator避免逐个任务重复传参。参数优先级Operator 参数覆盖连接元数据文档在 Reference 部分给出了一条关键规则直接传给SQLExecuteQueryOperator()的参数会覆盖 Airflow 连接元数据中对应的配置例如schema、login、password等。也就是说Operator 上的database参数若显式指定将优先于连接里配置的 Schema。这一点对多库作业很实用可以维护一个基础连接再在个别任务中临时切换到其他数据库而不必为每个库单独创建连接。SQLExecuteQueryOperator 完整参数说明从 SQLExecuteQueryOperator 源码文档 的 docstring 可以补充文档未逐一列出的参数含义参数说明sql要执行的 SQL 代码也可以是指向.sql模板文件的路径会经过模板渲染autocommit是否每条命令自动提交默认Falseparameters渲染 SQL 时使用的参数映射/可迭代对象handler应用于 cursor 的函数默认fetch_all_handleroutput_processor应用于结果的函数默认default_output_processorconn_id连接 IDdatabase覆盖连接中定义的数据库名split_statements是否拆分单条 SQL 字符串为多条语句默认沿用 Hookrun方法的默认值return_last是否只返回最后一条语句的结果默认Trueshow_return_value_in_logs是否把结果打印到任务日志默认False大数据集慎用requires_result_fetch为True时确保查询结果在执行完成前被取回模板化方面源码声明了template_fields (sql, parameters, ...)与template_ext (.sql, .json)sql.py即sql与parameters支持 Airflow 模板渲染SQL 也可外置为.sql文件。从execute()的调用链可以看出执行流程get_db_hook()解析出ImpalaHook→ 调用hook.run(sql..., autocommit..., parameters..., handler..., return_last...)→ 若开启 XCom pushdo_xcom_pushTrue则对结果执行_process_output并推入 XComexecute 方法。对于 Impala 来说ImpalaHook继承自DbApiHookrun、insert_rows、get_first、get_records、get_df等方法均可直接复用test_impala.py 中的测试验证了get_first/get_records/get_df支持 pandas 与 polars以及 Hook Lineagesend_sql_hook_lineage在 Impala 上均正常工作。小结与延伸阅读核心用法安装apache-airflow-providers-apache-impala配置impala类型连接Host / Port 21050 / Schema / Login / Password / Extra在 DAG 中使用SQLExecuteQueryOperator并通过conn_id引用该连接参数优先级Operator 显式参数如database覆盖连接元数据如schema认证扩展Kerberos 场景在连接 Extra 中配置auth_mechanism: GSSAPI需kerberosextraSSL 通过use_ssl控制源码验证入口ImpalaHook 实现、单元测试、系统测试示例 DAG、Provider 元信息更多 Impala SQL 语法细节请查阅 Apache Impala 官方文档。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow Impala Provider 实战：使用 SQLExecuteQueryOperator 连接与操作 Apache Impala
Apache Airflow Impala Provider 实战使用 SQLExecuteQueryOperator 连接与操作 Apache Impala【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本文基于 Airflow 仓库中 Impala Provider 的官方操作文档providers/apache/impala/docs/operators.rst展开讲解如何在 Airflow DAG 中通过SQLExecuteQueryOperator对 Apache Impala 集群执行 SQL 查询。读完本文你将掌握 Impala 连接的完整元数据配置、示例 DAG 的写法与参数优先级规则并能结合 ImpalaHook 源码 理解连接字段到impyla客户端的映射原理。背景为什么 Impala 使用通用的 SQLExecuteQueryOperatorImpala Provider 的官方文档明确指出此前 Impala 曾有专用的 Operator该专用 Operator 已被弃用deprecated现在应统一使用SQLExecuteQueryOperator来自airflow.providers.common.sql.operators.sql来对 Apache Impala 集群执行 SQL 查询。这一变化意味着 Impala 与其他数据库Postgres、MySQL、Hive 等在 Airflow 中的用法趋同——只要连接类型connection type对应的 Hook 实现正确通用的 SQL 系列 Operator 即可驱动它。从 provider.yaml 可以看到该 Provider 的元信息为包名apache-airflow-providers-apache-impala当前仓库中版本为 1.9.3lifecycle: production注册的 Hook 为 ImpalaHookPython 模块airflow.providers.apache.impala.hooks.impala注册的连接类型connection-type: impala绑定hook-class-name: airflow.providers.apache.impala.hooks.impala.ImpalaHook。也就是说当你使用impala类型的 Airflow 连接时Airflow 会自动解析出ImpalaHook而SQLExecuteQueryOperator则通过该 Hook 执行 SQL。安装 Provider 包与依赖文档中有一个重要提示必须安装apache-airflow-providers-apache-impala包才能启用 Impala 支持pip install apache-airflow-providers-apache-impala结合 pyproject.toml 与 README.rst该包的依赖与兼容性如下依赖版本要求impyla0.22.0,1.0apache-airflow-providers-common-sql1.32.0apache-airflow-providers-common-compat1.12.0apache-airflow2.11.0可选依赖extrasExtra依赖kerberoskerberos1.3.0Kerberos/GSSAPI 认证场景sqlalchemysqlalchemy1.4.54需要构建 SQLAlchemy URL 时使用该包支持 Python 3.10 3.14。注意impyla是实际的数据库客户端底层通过 HS2Hive Server 2协议与impalad通信连接参数如 SSL、认证机制都经由它传递。配置 Impala 连接Connection Metadata使用conn_id参数连接 Impala 实例时连接元数据Connection Metadata的结构如下完整继承自 operators.rst参数含义Host (string)Impala 守护进程的域名或 IP 地址可以是任意一个impalad服务节点Schema (string)默认数据库名可选。若为空行为由具体实现决定Login (string)认证用户名如适用例如 LDAP 用户Password (string)认证密码如适用Port (int)Impala 服务端口默认21050注意与 Hive 端口通常不同Extra (JSON)附加连接配置例如{use_ssl: false, auth: NOSASL}字段详细说明可参考 Provider 自带的连接文档 providers/apache/impala/docs/connections/impala.rst其中强调 Host/Port 对应 HS2 协议端点Impala 默认端口为21050Extra 字段是一个 JSON 字典其中的键会作为额外参数传给impyla连接。此外Impala 的 Hook 与 Operator 默认使用的连接 ID 为impala_default见 ImpalaHook 中的default_conn_name impala_default如果你按默认方式配置连接直接使用这个名字即可省去显式传conn_id。源码视角连接字段如何映射到 impylaImpalaHook 是DbApiHook的子类其get_conn()方法把 Airflow 连接对象的各字段逐一映射为impyla.dbapi.connect的调用参数def get_conn(self) - Connection: conn_id: str self.get_conn_id() connection self.get_connection(conn_id) return connect( hostconnection.host, portconnection.port, userconnection.login, passwordconnection.password, databaseconnection.schema, **connection.extra_dejson, # Extra JSON 中的键值原样展开为 connect() 参数 )两个值得注意的实现细节schema映射为databaselogin映射为user——配置连接时 Schema 字段就是 impyla 的默认数据库connection.extra_dejson被直接解包**因此 Extra 中写的任何 JSON 键如use_ssl、auth_mechanism、configuration都会透传给impyla的connect()。单元测试 test_impala.py 对此做了两类验证普通场景extra{use_ssl: True}最终断言为connect(host..., port21050, user..., password..., database..., use_sslTrue)Kerberos 场景extra{auth_mechanism: GSSAPI, use_ssl: True}最终断言为connect(..., use_sslTrue, auth_mechanismGSSAPI)。如果你在集群中启用了 Kerberos 认证需要在 Extra 中配置auth_mechanism: GSSAPI并安装kerberosextra。另外ImpalaHook还提供了sqlalchemy_url属性与get_uri()方法impala.py可将连接渲染为impala://user:passhost:21050/db?queryparams形式的 SQLAlchemy URL。该功能依赖可选的sqlalchemyextra——未安装时会抛出AirflowOptionalProviderFeatureException并提示以pip install apache-airflow-providers-apache-impala[sqlalchemy]安装。构建 URL 时若连接缺少host或login会抛出带明确信息的ValueError。示例在 DAG 中使用 SQLExecuteQueryOperator 连接 Impala官方文档通过exampleinclude指令引用了系统测试 DAG example_impala.py标记[START howto_operator_impala][END howto_operator_impala]区段。下面给出该示例 DAG 的完整可运行内容import datetime from airflow import DAG from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator DAG_ID example_impala with DAG( dag_idDAG_ID, start_datedatetime.datetime(2025, 1, 1), default_args{conn_id: my_impala_conn}, # Impala 连接 ID scheduleonce, catchupFalse, ) as dag: create_table_impala_task SQLExecuteQueryOperator( task_idcreate_table_impala, sql CREATE TABLE IF NOT EXISTS impala_example ( a STRING, b INT ) PARTITIONED BY (c INT) , ) alter_table_impala_task SQLExecuteQueryOperator( task_idalter_table_impala, sqlALTER TABLE impala_example ADD PARTITION (c1), ) insert_data_impala_task SQLExecuteQueryOperator( task_idinsert_data_impala_task, sqlINSERT INTO impala_example PARTITION (c1) VALUES (a, 1), (a, 2), (b, 3), ) select_data_impala_task SQLExecuteQueryOperator( task_idselect_data_impala, sqlSELECT * FROM impala_example, ) drop_table_impala_task SQLExecuteQueryOperator( task_iddrop_table_impala, sqlDROP TABLE impala_example, ) ( create_table_impala_task alter_table_impala_task insert_data_impala_task select_data_impala_task drop_table_impala_task )注上方代码中insert_data_impala_task的task_id已修正为不与变量名冲突的写法其余 SQL 与依赖链与原文件 example_impala.py 保持一致。示例覆盖了 Impala 的典型 DDL/DML 生命周期建分区表、加分区、按分区插入、查询、删表并且通过default_args把conn_id统一注入到所有 Operator避免逐个任务重复传参。参数优先级Operator 参数覆盖连接元数据文档在 Reference 部分给出了一条关键规则直接传给SQLExecuteQueryOperator()的参数会覆盖 Airflow 连接元数据中对应的配置例如schema、login、password等。也就是说Operator 上的database参数若显式指定将优先于连接里配置的 Schema。这一点对多库作业很实用可以维护一个基础连接再在个别任务中临时切换到其他数据库而不必为每个库单独创建连接。SQLExecuteQueryOperator 完整参数说明从 SQLExecuteQueryOperator 源码文档 的 docstring 可以补充文档未逐一列出的参数含义参数说明sql要执行的 SQL 代码也可以是指向.sql模板文件的路径会经过模板渲染autocommit是否每条命令自动提交默认Falseparameters渲染 SQL 时使用的参数映射/可迭代对象handler应用于 cursor 的函数默认fetch_all_handleroutput_processor应用于结果的函数默认default_output_processorconn_id连接 IDdatabase覆盖连接中定义的数据库名split_statements是否拆分单条 SQL 字符串为多条语句默认沿用 Hookrun方法的默认值return_last是否只返回最后一条语句的结果默认Trueshow_return_value_in_logs是否把结果打印到任务日志默认False大数据集慎用requires_result_fetch为True时确保查询结果在执行完成前被取回模板化方面源码声明了template_fields (sql, parameters, ...)与template_ext (.sql, .json)sql.py即sql与parameters支持 Airflow 模板渲染SQL 也可外置为.sql文件。从execute()的调用链可以看出执行流程get_db_hook()解析出ImpalaHook→ 调用hook.run(sql..., autocommit..., parameters..., handler..., return_last...)→ 若开启 XCom pushdo_xcom_pushTrue则对结果执行_process_output并推入 XComexecute 方法。对于 Impala 来说ImpalaHook继承自DbApiHookrun、insert_rows、get_first、get_records、get_df等方法均可直接复用test_impala.py 中的测试验证了get_first/get_records/get_df支持 pandas 与 polars以及 Hook Lineagesend_sql_hook_lineage在 Impala 上均正常工作。小结与延伸阅读核心用法安装apache-airflow-providers-apache-impala配置impala类型连接Host / Port 21050 / Schema / Login / Password / Extra在 DAG 中使用SQLExecuteQueryOperator并通过conn_id引用该连接参数优先级Operator 显式参数如database覆盖连接元数据如schema认证扩展Kerberos 场景在连接 Extra 中配置auth_mechanism: GSSAPI需kerberosextraSSL 通过use_ssl控制源码验证入口ImpalaHook 实现、单元测试、系统测试示例 DAG、Provider 元信息更多 Impala SQL 语法细节请查阅 Apache Impala 官方文档。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考