如何计算apache气流传感器的总执行时间答案

【问题标题】：How to calculate apache airflow sensor total execution time如何计算apache气流传感器的总执行时间
【发布时间】：2022-06-11 04:08:22
【问题描述】：

sensor_job = PythonSensor(
            task_id='sensor_id',
            python_callable=call_jobsensor,
            poke_interval=10,
            timeout=7 * 60,
            mode='reschedule',
        )        


def call_jobsensor():
    # start timer
    # do something
    # stop timer

用例 - 在这个例子中，我试图捕捉完成传感器工作所花费的总时间。此处完成意味着它应该超时或传感器返回 true。如果我用传统方法，它会以小块的形式返回时间信息（每次传感器调用这个函数），但我需要的是总时间。

我可以在这里使用 Airflow 或 Statsd 的任何方法或支持吗？还是有其他建议？

【问题讨论】：

标签： airflow sensors statsd

【解决方案1】：

假设您使用的是 Airflow v2：

您可以使用气流数据库后端来获取 DAG 中任务的持续时间。

这假设您已经设置了一个 airflow database backend 并使用您的主机、架构、用户名和密码配置了默认的 airflow_db 连接。

根据您使用的数据库，您需要安装相关的提供程序。

您可以查看apache-airflow-providers-mysql here 的文档。

还有apache-airflow-providers-postgres here 的文档。

您需要将提供程序全局安装到气流环境。

使用 PostgreSQL 访问的 DAG 示例是（如果使用 MySQL，只需将导入交换为 MySQLHook）：

import pendulum

# from airflow.providers.mysql.hooks.mysql import MySqlHook  # install apache-airflow-providers-mysql
from airflow.providers.postgres.hooks.postgres import PostgresHook  # install apache-airflow-providers-postgres
from airflow.decorators import dag, task


@dag(start_date=pendulum.yesterday(tz="Europe/London"))
def test_dag():
    @task()
    def test_duration_task():
        from time import sleep
        sleep(3)
        
    @task()
    def run_query(**context):
        postgres_hook = PostgresHook(postgres_conn_id="airflow_db")
        records = postgres_hook.get_records(sql="select duration from task_instance where task_id='test_duration_task' order by end_date desc")
        print(f"task duration for test_duration_task = {records[0][0]}")

    test_duration_task() >> run_query()


test_dag_failure_dag = test_dag()

run_query的日志输出为：

[2022-06-10, 19:57:59 UTC] {base.py:68} INFO - Using connection ID '***_db' for task execution.
[2022-06-10, 19:57:59 UTC] {logging_mixin.py:115} INFO - task duration for test_duration_task = 3.200903
[2022-06-10, 19:57:59 UTC] {python.py:173} INFO - Done. Returned value was: None

【讨论】：