获取 apache airflow 任务的 unique_id

Question

我是airflow的新手。在我的ETL管道公司中，目前我们正在使用Crontab和自定义调度程序（内部开发）。现在我们计划为我们所有的数据管道场景实现apache airflow。为此，在探索功能的同时无法找到每个任务实例/Dag 的unique_id。当我搜索大多数解决方案时，它们都以宏和模板结束。但是它们都没有为任务提供 uniqueID 。但是我能够在 UI 中看到增量 uniqueID对于每个任务。有什么方法可以轻松访问我的 python 方法中的这些变量。主要用例是我需要将这些 ID 作为参数传递给 Python/ruby/Pentaho 作业，称为脚本/方法。

例如

我的 shell 脚本“test.sh”需要两个参数，一个是 run_id，另一个是 collection_id。目前，我们正在从集中数据库生成这个唯一的 run_id 并将其传递给作业。如果它已经存在于气流上下文中，我们将使用它

from airflow.operators.bash_operator import BashOperator
from datetime import date, datetime, timedelta
from airflow import DAG

shell_command =  "/data2/test.sh -r run_id -c collection_id"


putfiles_s3 = BashOperator(
                task_id='putfiles_s3',
                bash_command=shell_command,
                dag=dag)

在执行此 Dag（计划/手动）时为每次运行寻找唯一的 run_id（Dag 级别/任务级别）

注意：这是一个示例任务。该 Dag 将会有多个依赖任务。附加 Airflow UI 中的 Job_Id 屏幕截图

谢谢阿努普R

Answer 1

{{ ti.job_id }}是你想要的：

from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator
from airflow import DAG


dag = DAG(
    "job_id",
    start_date=datetime(2018, 1, 1),
)

with dag:
    BashOperator(
        task_id='unique_id',
        bash_command="echo {{ ti.job_id }}",
    )

这将在运行时有效。此执行的日志如下所示：

[2018-01-03 10:28:37,523] {bash_operator.py:80} INFO - Temporary script location: /tmp/airflowtmpcj0omuts//tmp/airflowtmpcj0omuts/unique_iddq7kw0yj  
[2018-01-03 10:28:37,524] {bash_operator.py:88} INFO - Running command: echo 4
[2018-01-03 10:28:37,621] {bash_operator.py:97} INFO - Output:
[2018-01-03 10:28:37,648] {bash_operator.py:101} INFO - 4

请注意，这仅在运行时有效，因此 WebUI 中的“渲染模板”视图将显示 None 而不是数字。