如何获取气流 dag 运行的 JobID？答案

【问题标题】：How to get the JobID for the airflow dag runs?如何获取气流 dag 运行的 JobID？
【发布时间】：2017-09-06 20:16:02
【问题描述】：

当我们执行 dagrun 时，在 Airflow UI 的“图表视图”中，我们会获得每个作业运行的详细信息。

JobID 类似于“scheduled__2017-04-11T10:47:00”。

我需要这个 JobID 来跟踪和创建日志，我在其中维护每个任务/dagrun 所花费的时间。

所以我的问题是如何在运行的同一个 dag 中获取 JobID。

谢谢，车坦

【问题讨论】：

标签： python pyspark airflow apache-airflow

【解决方案1】：

这个值实际上叫做run_id，可以通过上下文或者宏来访问。

在 python 运算符中，这是通过上下文访问的，而在 bash 运算符中，这是通过 bash_command 字段上的 jinja 模板访问的。

from airflow.models import DAG
from datetime import datetime
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator


dag = DAG(
    dag_id='run_id',
    schedule_interval=None,
    start_date=datetime(2017, 2, 26)
)

def my_func(**kwargs):
    context = kwargs
    print(context['dag_run'].run_id)

t1 = PythonOperator(
    task_id='python_run_id',
    python_callable=my_func,
    provide_context=True,
    dag=dag
    )

t2 = BashOperator(
    task_id='bash_run_id',
    bash_command='echo {{run_id}}',
    dag=dag)

t1.set_downstream(t2)

以这个 dag 为例，检查每个操作员的日志，你应该会看到在日志中打印了run_id。

【讨论】：

这仅适用于计划运行吗？我从 CLI 运行并得到 kwargs['dag_run'] is None
也适用于 SSHOperator :t3 = SSHOperator( dag=dag, ssh_conn_id='some-great-worker', task_id='666', command=f"""{{{{ run_id } }}}""")