Apache Airflow 集成 ArangoDBConnection 配置完全指南与 ArangoDBHook 源码级解析【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本指南围绕 Apache Airflow 官方 ArangoDB Provider 的 Connection 配置展开完整讲解连接 ArangoDB 所需的四个必填字段Host、Database/Schema、Username、Password并结合仓库源码深入剖析ArangoDBHook的字段映射、集群多 Coordinator 支持、UI 表单行为以及基于该连接工作的AQLOperator、AQLSensor与ArangoDBCollectionOperator。读完本文你将能够在 Airflow 中正确创建 ArangoDB 连接并将其应用到 DAG 的查询、监控与集合操作任务中。1. ArangoDB Connection 是什么ArangoDB 是一种支持文档、图与键值三种数据模型的多模型数据库其查询语言为 AQLArangoDB Query Language。Apache Airflow 通过apache-airflow-providers-arangodbProvider 包与 ArangoDB 交互而ArangoDB Connection 正是为这种交互提供凭据credentials与连接入口的配置载体——它保存了访问 ArangoDB 所需的地址、数据库、用户名和密码供 Hook、Operator 与 Sensor 统一读取使用。原文档 providers/arangodb/docs/connections/arangodb.rst 明确指出The ArangoDB connection provides credentials for accessing the ArangoDB.ArangoDB Connection 提供访问 ArangoDB 的凭据。这是该 Provider 所有任务的基础设施无论是执行 AQL 查询的AQLOperator还是等待数据出现的AQLSensor最终都会通过ArangoDBHook读取这条连接来建立与数据库的会话。2. 配置 ArangoDB Connection四个必填字段根据原文档配置 ArangoDB Connection 时需要在 Airflow 的 Connection 管理界面Admin → Connections中填写以下字段四个字段全部为必填2.1 ArangoDB Host必填Specify ArangoDB Host URL or comma separated list of URLs (coordinators in a cluster), e.g.http://127.0.0.1:8529orhttp://127.0.0.1:8529,http://127.0.0.1:8530.单机模式填写单个 ArangoDB 服务 URL例如http://127.0.0.1:85298529 是 ArangoDB 的默认 HTTP 端口。集群模式可填写逗号分隔的多个 URL 列表用于指定集群中的多个 coordinator 节点例如http://127.0.0.1:8529,http://127.0.0.1:8530。从源码看Host 字段在 Hook 中的消费逻辑位于 providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.pyproperty def hosts(self) - list[str]: if not self._conn.host: raise AirflowException(fNo ArangoDB Host(s) provided in connection: {self.arangodb_conn_id!r}.) return self._conn.host.split(,)可见 Hook 会将 Host 字段按逗号,拆分成hosts列表再传给ArangoDBClient(hostsself.hosts)。这意味着多 coordinator URL 的写法最终会映射为 python-arango 客户端的多主机列表由客户端自行进行负载均衡与故障切换。同时若未填写 HostHook 会抛出AirflowException(No ArangoDB Host(s) provided ...)印证了该字段的必填属性。2.2 ArangoDB Database/Schema必填SpecifyDatabase/Schemafor the ArangoDB. eg._system.该字段对应 Airflow Connection 的Schema列。ArangoDB 预置了一个名为_system的系统数据库作为默认示例值。Hook 中该字段的读取逻辑为 arangodb.py#L80-L84property def database(self) - str: if not self._conn.schema: raise AirflowException(fNo ArangoDB Database provided in connection: {self.arangodb_conn_id!r}.) return self._conn.schema即 Airflow Connection 的schema属性被映射为 ArangoDB 的 database 名称最终在db_conn中调用self.client.db(nameself.database, usernameself.username, passwordself.password)完成对指定数据库的连接。如果留空同样会抛出AirflowException明确其为必填项。2.3 ArangoDB Username必填Specifyusernamefor the ArangoDB, e.g.root.对应 Airflow Connection 的Login列示例默认值为rootArangoDB 的超级用户。Hook 中的映射见 arangodb.py#L86-L90property def username(self) - str: if not self._conn.login: raise AirflowException(fNo ArangoDB Username provided in connection: {self.arangodb_conn_id!r}.) return self._conn.login2.4 ArangoDB Password必填Specifypasswordfor the ArangoDB.对应 Airflow Connection 的Password列。与前三者不同密码字段在 Hook 中允许为空字符串不会触发异常见 arangodb.py#L92-L94property def password(self) - str: return self._conn.password or 不过在文档语义上它仍属必填因为正常连接 ArangoDB 需要有效凭据此处的宽松处理仅为避免无密码环境下的解析报错。连接时密码与用户名、数据库名一起传入client.db(...)。2.5 字段映射速查表以下为原文档四个字段与 Airflow Connection 表单列、Hook 属性、底层调用的完整映射文档字段Airflow Connection 列Hook 属性底层使用位置是否必填ArangoDB HostHosthosts按逗号拆分ArangoDBClient(hosts...)是ArangoDB Database/SchemaSchemadatabaseclient.db(name...)是ArangoDB UsernameLoginusernameclient.db(username...)是ArangoDB PasswordPasswordpasswordclient.db(password...)是3. 创建 Connection 的两种方式3.1 通过 Web UI 创建在 Airflow Web UI 的Admin → Connections页面点击新增连接选择类型Conn Type为ArangoDB。得益于ArangoDBHook.get_ui_field_behaviour()的定义arangodb.py#L191-L208表单会呈现以下定制行为隐藏字段port端口与extra附加参数两个字段被隐藏无需填写字段重命名Host 显示为 ArangoDB Host URL or comma separated list of URLs (coordinators in a cluster)Schema 显示为 ArangoDB DatabaseLogin 显示为 ArangoDB UsernamePassword 显示为 ArangoDB Password占位符示例Host 显示eg.http://127.0.0.1:8529 or http://127.0.0.1:8529,http://127.0.0.1:8530 (coordinators in a cluster)Schema 显示_systemLogin 显示rootPassword 显示password。这些 UI 行为与 provider.yaml 中connection-types段的ui-field-behaviour声明完全一致两处共同保证了 Web 表单对用户的引导性。3.2 通过代码 / CLI 创建连接信息本质上就是一条 AirflowConnection记录。仓库测试 providers/arangodb/tests/unit/arangodb/hooks/test_arangodb.py#L30-L42 展示了该连接在测试环境中的标准构造方式可直接作为代码方式创建连接的参考Connection( conn_idarangodb_default, conn_typearangodb, hosthttp://127.0.0.1:8529, loginroot, passwordpassword, schema_system, )其中conn_type必须为arangodbconn_id默认为arangodb_default这是ArangoDBHook的默认连接 ID见 arangodb.py#L51-L54。生产环境中更推荐使用 Airflow 的airflow connections add命令或环境变量 / 密钥后端来管理该连接避免明文写死在 DAG 中。4. Hook 层工作原理连接如何被消费配置好的 Connection 由ArangoDBHook定义于 providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.py统一消费。其核心成员如下conn_name_attr arangodb_conn_id、default_conn_name arangodb_default默认连接 ID 机制conn_type arangodb对应 Connection 记录中的 conn_typehook_name ArangoDB在 UI 中显示的名称。连接建立流程为三级cached_property链路clientArangoDBClient(hostsself.hosts)创建并缓存基于 python-arango 的客户端实例db_connself.client.db(nameself.database, usernameself.username, passwordself.password)用连接中的数据库名、用户名、密码打开指定数据库返回StandardDatabase数据库 API 包装器_connself.get_connection(self.arangodb_conn_id)从 Airflow 元数据库读取 Connection 记录。因为三者均为cached_property在同一 Hook 实例生命周期内只会建立一次连接后续调用直接复用避免重复建连开销。在数据库操作层面Hook 还提供了一系列文档/集合/数据库管理方法query(query, **kwargs)在会话中执行 AQL 查询返回Cursor结果集若执行失败或返回类型非Cursor会抛出AirflowExceptionarangodb.py#L100-L116create_collection/delete_collection先通过has_collection判断存在性再进行创建或删除并返回布尔值表示是否真正执行了操作create_database/create_graph同理管理数据库与图insert_documents/update_documents/replace_documents/delete_documents对集合执行批量文档写入、更新、替换与删除均使用silentTrue并通过DocumentInsertError等异常类型捕获错误。这些方法与 python-arango 官方客户端 API 一一对应是 Operator 层所有集合操作的地基。5. 基于该 Connection 的 Operator 与 Sensor 实战Connection 配置完成后即可在 DAG 中通过arangodb_conn_id参数引用。仓库提供了完整的示例 DAGproviders/arangodb/src/airflow/providers/arangodb/example_dags/example_arangodb.py下面逐一展开。5.1 AQLOperator执行 AQL 查询AQLOperatorproviders/arangodb/src/airflow/providers/arangodb/operators/arangodb.py#L30-L66在 ArangoDB 数据库中执行 AQL 查询operator AQLOperator( task_idaql_operator, queryFOR doc IN students RETURN doc, dagdag, result_processorlambda cursor: print([document[name] for document in cursor]), )关键参数query要执行的 AQL 语句必填。它同时是模板字段template_fields (query,)支持 Jinja 模板渲染arangodb_conn_id引用 ArangoDB Connection默认为arangodb_defaultresult_processor可选的 Callable用于对 ArangoDB 返回的Cursor结果做进一步处理。执行时Operator 会实例化ArangoDBHook(arangodb_conn_id...)并调用hook.query(self.query)若提供了result_processor则把结果 Cursor 交给它处理。单元测试 providers/arangodb/tests/unit/arangodb/operators/test_arangodb.py#L26-L33 验证了AQLOperator.execute会用默认连接 ID 实例化 Hook 并恰好调用一次query。5.2 ArangoDBCollectionOperator集合级批量操作ArangoDBCollectionOperatorarangodb.py#L69-L151在同一任务内对指定集合执行文档批量操作collection_name目标集合名必填documents_to_insert/documents_to_update/documents_to_replace/documents_to_delete均为list[dict]文档列表可组合使用delete_collection布尔值为True时删除整个集合。其execute逻辑会先校验五个操作参数中至少指定一个否则抛出ValueError(At least one operation must be specified.)对应测试 test_arangodb.py#L53-L60随后按 insert → update → replace → delete → delete_collection 的顺序依次调用 Hook 的对应方法。示例ArangoDBCollectionOperator( task_idinsert_task, collection_namestudents, documents_to_insert[{_key: lola, first: Lola, last: Martin}], )5.3 AQLSensor等待查询结果出现AQLSensorproviders/arangodb/src/airflow/providers/arangodb/sensors/arangodb.py#L30-L55周期性地执行 AQL 查询并检查结果集是否非空直到满足条件或超时sensor AQLSensor( task_idaql_sensor, queryFOR doc IN students FILTER doc.name judy RETURN doc, timeout60, poke_interval10, dagdag, )其poke方法执行hook.query(self.query, countTrue).count()获取记录数返回records ! 0配合BaseSensorOperator自带的timeout超时秒数与poke_interval轮询间隔秒数参数即可实现等待 students 集合中出现姓名为 judy 的文档这类典型场景。5.4 使用 .sql 模板文件加载查询AQLOperator与AQLSensor都支持通过template_ext (.sql,)从.sql文件中加载查询语句避免在 Python 代码中拼接长 AQL。示例 DAG 中展示了两种写法operator2 AQLOperator( task_idaql_operator_template_file, dagdag, result_processorlambda cursor: print([document[name] for document in cursor]), querysearch_all.sql, ) sensor2 AQLSensor( task_idaql_sensor_template_file, querysearch_judy.sql, timeout60, poke_interval10, dagdag, )注意.sql文件的路径默认相对于dags/目录解析若文件放在其他位置需要在创建DAG对象时通过template_searchpath参数指定搜索路径。同时template_fields_renderers {query: sql}声明了该模板字段按 SQL 语法高亮渲染。6. 环境要求与安装使用 ArangoDB Connection 前需要先安装 Provider 包并满足依赖版本要求见 providers/arangodb/README.rst 与 providers/arangodb/pyproject.tomlpip install apache-airflow-providers-arangodb依赖约束依赖包版本要求apache-airflow2.11.0apache-airflow-providers-common-compat1.10.1python-arango7.3.2Provider 支持 Python 3.10 ~ 3.14。当前仓库中该 Provider 的最新发布版本为 2.9.6见 provider.yaml 的版本列表状态为ready、生命周期为production可安全用于生产环境。7. 常见问题与排错建议No ArangoDB Host(s) provided in connectionConnection 的 Host 字段未填写。按文档要求补全http://host:8529格式的 URL集群场景请用逗号分隔多个 coordinator URL。No ArangoDB Database provided in connectionSchema 字段未填写补上目标数据库名如_system。No ArangoDB Username provided in connectionLogin 字段未填写。AQL 执行失败AQLOperator或AQLSensor抛出的Failed to execute AQLQuery, error: ...来自 Hook 的query方法对AQLQueryExecuteError的捕获与包装arangodb.py#L115-L116请检查 AQL 语法、目标集合是否存在以及连接中的数据库名是否正确。.sql文件找不到确认查询文件位于 dags/ 目录下或已通过 DAG 的template_searchpath指定了正确路径。8. 总结ArangoDB Connection 是 Apache Airflow 与 ArangoDB 集成的凭据枢纽其四个必填字段Host、Database/Schema、Username、Password通过ArangoDBHook精确映射为 python-arango 客户端的连接参数其中 Host 支持逗号分隔的多 coordinator URL 以适配集群部署。掌握了 Connection 的配置再配合AQLOperator、AQLSensor与ArangoDBCollectionOperator即可在 DAG 中完成 AQL 查询、结果处理、集合批量增删改与数据就绪感知的完整工作流编排。延伸阅读仓库内资源连接配置原文档providers/arangodb/docs/connections/arangodb.rstHook 实现providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.pyOperator 与 Sensor 实现operators/arangodb.py、sensors/arangodb.py可运行的示例 DAGexample_dags/example_arangodb.py单元测试hooks/test_arangodb.py、operators/test_arangodb.pyProvider 元数据与连接类型声明provider.yaml【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow 集成 ArangoDB：Connection 配置完全指南与 ArangoDBHook 源码级解析
Apache Airflow 集成 ArangoDBConnection 配置完全指南与 ArangoDBHook 源码级解析【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本指南围绕 Apache Airflow 官方 ArangoDB Provider 的 Connection 配置展开完整讲解连接 ArangoDB 所需的四个必填字段Host、Database/Schema、Username、Password并结合仓库源码深入剖析ArangoDBHook的字段映射、集群多 Coordinator 支持、UI 表单行为以及基于该连接工作的AQLOperator、AQLSensor与ArangoDBCollectionOperator。读完本文你将能够在 Airflow 中正确创建 ArangoDB 连接并将其应用到 DAG 的查询、监控与集合操作任务中。1. ArangoDB Connection 是什么ArangoDB 是一种支持文档、图与键值三种数据模型的多模型数据库其查询语言为 AQLArangoDB Query Language。Apache Airflow 通过apache-airflow-providers-arangodbProvider 包与 ArangoDB 交互而ArangoDB Connection 正是为这种交互提供凭据credentials与连接入口的配置载体——它保存了访问 ArangoDB 所需的地址、数据库、用户名和密码供 Hook、Operator 与 Sensor 统一读取使用。原文档 providers/arangodb/docs/connections/arangodb.rst 明确指出The ArangoDB connection provides credentials for accessing the ArangoDB.ArangoDB Connection 提供访问 ArangoDB 的凭据。这是该 Provider 所有任务的基础设施无论是执行 AQL 查询的AQLOperator还是等待数据出现的AQLSensor最终都会通过ArangoDBHook读取这条连接来建立与数据库的会话。2. 配置 ArangoDB Connection四个必填字段根据原文档配置 ArangoDB Connection 时需要在 Airflow 的 Connection 管理界面Admin → Connections中填写以下字段四个字段全部为必填2.1 ArangoDB Host必填Specify ArangoDB Host URL or comma separated list of URLs (coordinators in a cluster), e.g.http://127.0.0.1:8529orhttp://127.0.0.1:8529,http://127.0.0.1:8530.单机模式填写单个 ArangoDB 服务 URL例如http://127.0.0.1:85298529 是 ArangoDB 的默认 HTTP 端口。集群模式可填写逗号分隔的多个 URL 列表用于指定集群中的多个 coordinator 节点例如http://127.0.0.1:8529,http://127.0.0.1:8530。从源码看Host 字段在 Hook 中的消费逻辑位于 providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.pyproperty def hosts(self) - list[str]: if not self._conn.host: raise AirflowException(fNo ArangoDB Host(s) provided in connection: {self.arangodb_conn_id!r}.) return self._conn.host.split(,)可见 Hook 会将 Host 字段按逗号,拆分成hosts列表再传给ArangoDBClient(hostsself.hosts)。这意味着多 coordinator URL 的写法最终会映射为 python-arango 客户端的多主机列表由客户端自行进行负载均衡与故障切换。同时若未填写 HostHook 会抛出AirflowException(No ArangoDB Host(s) provided ...)印证了该字段的必填属性。2.2 ArangoDB Database/Schema必填SpecifyDatabase/Schemafor the ArangoDB. eg._system.该字段对应 Airflow Connection 的Schema列。ArangoDB 预置了一个名为_system的系统数据库作为默认示例值。Hook 中该字段的读取逻辑为 arangodb.py#L80-L84property def database(self) - str: if not self._conn.schema: raise AirflowException(fNo ArangoDB Database provided in connection: {self.arangodb_conn_id!r}.) return self._conn.schema即 Airflow Connection 的schema属性被映射为 ArangoDB 的 database 名称最终在db_conn中调用self.client.db(nameself.database, usernameself.username, passwordself.password)完成对指定数据库的连接。如果留空同样会抛出AirflowException明确其为必填项。2.3 ArangoDB Username必填Specifyusernamefor the ArangoDB, e.g.root.对应 Airflow Connection 的Login列示例默认值为rootArangoDB 的超级用户。Hook 中的映射见 arangodb.py#L86-L90property def username(self) - str: if not self._conn.login: raise AirflowException(fNo ArangoDB Username provided in connection: {self.arangodb_conn_id!r}.) return self._conn.login2.4 ArangoDB Password必填Specifypasswordfor the ArangoDB.对应 Airflow Connection 的Password列。与前三者不同密码字段在 Hook 中允许为空字符串不会触发异常见 arangodb.py#L92-L94property def password(self) - str: return self._conn.password or 不过在文档语义上它仍属必填因为正常连接 ArangoDB 需要有效凭据此处的宽松处理仅为避免无密码环境下的解析报错。连接时密码与用户名、数据库名一起传入client.db(...)。2.5 字段映射速查表以下为原文档四个字段与 Airflow Connection 表单列、Hook 属性、底层调用的完整映射文档字段Airflow Connection 列Hook 属性底层使用位置是否必填ArangoDB HostHosthosts按逗号拆分ArangoDBClient(hosts...)是ArangoDB Database/SchemaSchemadatabaseclient.db(name...)是ArangoDB UsernameLoginusernameclient.db(username...)是ArangoDB PasswordPasswordpasswordclient.db(password...)是3. 创建 Connection 的两种方式3.1 通过 Web UI 创建在 Airflow Web UI 的Admin → Connections页面点击新增连接选择类型Conn Type为ArangoDB。得益于ArangoDBHook.get_ui_field_behaviour()的定义arangodb.py#L191-L208表单会呈现以下定制行为隐藏字段port端口与extra附加参数两个字段被隐藏无需填写字段重命名Host 显示为 ArangoDB Host URL or comma separated list of URLs (coordinators in a cluster)Schema 显示为 ArangoDB DatabaseLogin 显示为 ArangoDB UsernamePassword 显示为 ArangoDB Password占位符示例Host 显示eg.http://127.0.0.1:8529 or http://127.0.0.1:8529,http://127.0.0.1:8530 (coordinators in a cluster)Schema 显示_systemLogin 显示rootPassword 显示password。这些 UI 行为与 provider.yaml 中connection-types段的ui-field-behaviour声明完全一致两处共同保证了 Web 表单对用户的引导性。3.2 通过代码 / CLI 创建连接信息本质上就是一条 AirflowConnection记录。仓库测试 providers/arangodb/tests/unit/arangodb/hooks/test_arangodb.py#L30-L42 展示了该连接在测试环境中的标准构造方式可直接作为代码方式创建连接的参考Connection( conn_idarangodb_default, conn_typearangodb, hosthttp://127.0.0.1:8529, loginroot, passwordpassword, schema_system, )其中conn_type必须为arangodbconn_id默认为arangodb_default这是ArangoDBHook的默认连接 ID见 arangodb.py#L51-L54。生产环境中更推荐使用 Airflow 的airflow connections add命令或环境变量 / 密钥后端来管理该连接避免明文写死在 DAG 中。4. Hook 层工作原理连接如何被消费配置好的 Connection 由ArangoDBHook定义于 providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.py统一消费。其核心成员如下conn_name_attr arangodb_conn_id、default_conn_name arangodb_default默认连接 ID 机制conn_type arangodb对应 Connection 记录中的 conn_typehook_name ArangoDB在 UI 中显示的名称。连接建立流程为三级cached_property链路clientArangoDBClient(hostsself.hosts)创建并缓存基于 python-arango 的客户端实例db_connself.client.db(nameself.database, usernameself.username, passwordself.password)用连接中的数据库名、用户名、密码打开指定数据库返回StandardDatabase数据库 API 包装器_connself.get_connection(self.arangodb_conn_id)从 Airflow 元数据库读取 Connection 记录。因为三者均为cached_property在同一 Hook 实例生命周期内只会建立一次连接后续调用直接复用避免重复建连开销。在数据库操作层面Hook 还提供了一系列文档/集合/数据库管理方法query(query, **kwargs)在会话中执行 AQL 查询返回Cursor结果集若执行失败或返回类型非Cursor会抛出AirflowExceptionarangodb.py#L100-L116create_collection/delete_collection先通过has_collection判断存在性再进行创建或删除并返回布尔值表示是否真正执行了操作create_database/create_graph同理管理数据库与图insert_documents/update_documents/replace_documents/delete_documents对集合执行批量文档写入、更新、替换与删除均使用silentTrue并通过DocumentInsertError等异常类型捕获错误。这些方法与 python-arango 官方客户端 API 一一对应是 Operator 层所有集合操作的地基。5. 基于该 Connection 的 Operator 与 Sensor 实战Connection 配置完成后即可在 DAG 中通过arangodb_conn_id参数引用。仓库提供了完整的示例 DAGproviders/arangodb/src/airflow/providers/arangodb/example_dags/example_arangodb.py下面逐一展开。5.1 AQLOperator执行 AQL 查询AQLOperatorproviders/arangodb/src/airflow/providers/arangodb/operators/arangodb.py#L30-L66在 ArangoDB 数据库中执行 AQL 查询operator AQLOperator( task_idaql_operator, queryFOR doc IN students RETURN doc, dagdag, result_processorlambda cursor: print([document[name] for document in cursor]), )关键参数query要执行的 AQL 语句必填。它同时是模板字段template_fields (query,)支持 Jinja 模板渲染arangodb_conn_id引用 ArangoDB Connection默认为arangodb_defaultresult_processor可选的 Callable用于对 ArangoDB 返回的Cursor结果做进一步处理。执行时Operator 会实例化ArangoDBHook(arangodb_conn_id...)并调用hook.query(self.query)若提供了result_processor则把结果 Cursor 交给它处理。单元测试 providers/arangodb/tests/unit/arangodb/operators/test_arangodb.py#L26-L33 验证了AQLOperator.execute会用默认连接 ID 实例化 Hook 并恰好调用一次query。5.2 ArangoDBCollectionOperator集合级批量操作ArangoDBCollectionOperatorarangodb.py#L69-L151在同一任务内对指定集合执行文档批量操作collection_name目标集合名必填documents_to_insert/documents_to_update/documents_to_replace/documents_to_delete均为list[dict]文档列表可组合使用delete_collection布尔值为True时删除整个集合。其execute逻辑会先校验五个操作参数中至少指定一个否则抛出ValueError(At least one operation must be specified.)对应测试 test_arangodb.py#L53-L60随后按 insert → update → replace → delete → delete_collection 的顺序依次调用 Hook 的对应方法。示例ArangoDBCollectionOperator( task_idinsert_task, collection_namestudents, documents_to_insert[{_key: lola, first: Lola, last: Martin}], )5.3 AQLSensor等待查询结果出现AQLSensorproviders/arangodb/src/airflow/providers/arangodb/sensors/arangodb.py#L30-L55周期性地执行 AQL 查询并检查结果集是否非空直到满足条件或超时sensor AQLSensor( task_idaql_sensor, queryFOR doc IN students FILTER doc.name judy RETURN doc, timeout60, poke_interval10, dagdag, )其poke方法执行hook.query(self.query, countTrue).count()获取记录数返回records ! 0配合BaseSensorOperator自带的timeout超时秒数与poke_interval轮询间隔秒数参数即可实现等待 students 集合中出现姓名为 judy 的文档这类典型场景。5.4 使用 .sql 模板文件加载查询AQLOperator与AQLSensor都支持通过template_ext (.sql,)从.sql文件中加载查询语句避免在 Python 代码中拼接长 AQL。示例 DAG 中展示了两种写法operator2 AQLOperator( task_idaql_operator_template_file, dagdag, result_processorlambda cursor: print([document[name] for document in cursor]), querysearch_all.sql, ) sensor2 AQLSensor( task_idaql_sensor_template_file, querysearch_judy.sql, timeout60, poke_interval10, dagdag, )注意.sql文件的路径默认相对于dags/目录解析若文件放在其他位置需要在创建DAG对象时通过template_searchpath参数指定搜索路径。同时template_fields_renderers {query: sql}声明了该模板字段按 SQL 语法高亮渲染。6. 环境要求与安装使用 ArangoDB Connection 前需要先安装 Provider 包并满足依赖版本要求见 providers/arangodb/README.rst 与 providers/arangodb/pyproject.tomlpip install apache-airflow-providers-arangodb依赖约束依赖包版本要求apache-airflow2.11.0apache-airflow-providers-common-compat1.10.1python-arango7.3.2Provider 支持 Python 3.10 ~ 3.14。当前仓库中该 Provider 的最新发布版本为 2.9.6见 provider.yaml 的版本列表状态为ready、生命周期为production可安全用于生产环境。7. 常见问题与排错建议No ArangoDB Host(s) provided in connectionConnection 的 Host 字段未填写。按文档要求补全http://host:8529格式的 URL集群场景请用逗号分隔多个 coordinator URL。No ArangoDB Database provided in connectionSchema 字段未填写补上目标数据库名如_system。No ArangoDB Username provided in connectionLogin 字段未填写。AQL 执行失败AQLOperator或AQLSensor抛出的Failed to execute AQLQuery, error: ...来自 Hook 的query方法对AQLQueryExecuteError的捕获与包装arangodb.py#L115-L116请检查 AQL 语法、目标集合是否存在以及连接中的数据库名是否正确。.sql文件找不到确认查询文件位于 dags/ 目录下或已通过 DAG 的template_searchpath指定了正确路径。8. 总结ArangoDB Connection 是 Apache Airflow 与 ArangoDB 集成的凭据枢纽其四个必填字段Host、Database/Schema、Username、Password通过ArangoDBHook精确映射为 python-arango 客户端的连接参数其中 Host 支持逗号分隔的多 coordinator URL 以适配集群部署。掌握了 Connection 的配置再配合AQLOperator、AQLSensor与ArangoDBCollectionOperator即可在 DAG 中完成 AQL 查询、结果处理、集合批量增删改与数据就绪感知的完整工作流编排。延伸阅读仓库内资源连接配置原文档providers/arangodb/docs/connections/arangodb.rstHook 实现providers/arangodb/src/airflow/providers/arangodb/hooks/arangodb.pyOperator 与 Sensor 实现operators/arangodb.py、sensors/arangodb.py可运行的示例 DAGexample_dags/example_arangodb.py单元测试hooks/test_arangodb.py、operators/test_arangodb.pyProvider 元数据与连接类型声明provider.yaml【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考