运算符¶
您可以在 ArangoDBHook
中构建您自己的 Operator hook。
使用 AQLOperator
在 ArangoDB 中执行 AQL 查询。
您可以使用 AQLOperator
进一步处理您的结果,并使用 result_processor
Callable 按您的需求进一步处理结果。
列出 students 集合中所有文档的示例可以按如下方式实现
src/airflow/providers/arangodb/example_dags/example_arangodb.py
operator = AQLOperator(
task_id="aql_operator",
query="FOR doc IN students RETURN doc",
dag=dag,
result_processor=lambda cursor: print([document["name"] for document in cursor]),
)
您还可以提供文件模板 (.sql) 来加载查询,请记住路径是相对于 dags/ 文件夹的,如果您想提供任何其他路径,请在创建 DAG 对象时提供 template_searchpath,
src/airflow/providers/arangodb/example_dags/example_arangodb.py
operator2 = AQLOperator(
task_id="aql_operator_template_file",
dag=dag,
result_processor=lambda cursor: print([document["name"] for document in cursor]),
query="search_all.sql",
)
传感器¶
使用 AQLSensor
在 ArangoDB 中使用 AQL 查询等待文档或集合。
在 students 集合中等待学生姓名 judy 的文档的示例可以按如下方式实现
src/airflow/providers/arangodb/example_dags/example_arangodb.py
sensor = AQLSensor(
task_id="aql_sensor",
query="FOR doc IN students FILTER doc.name == 'judy' RETURN doc",
timeout=60,
poke_interval=10,
dag=dag,
)
类似于 AQLOperator,您还可以提供文件模板来加载查询 -
src/airflow/providers/arangodb/example_dags/example_arangodb.py
sensor2 = AQLSensor(
task_id="aql_sensor_template_file",
query="search_judy.sql",
timeout=60,
poke_interval=10,
dag=dag,
)