获取 apache 气流任务的 unique_id

Question

我是气流新手。在我的 ETL 管道公司中，目前我们正在使用 Crontab 和自定义调度程序（内部开发）。现在我们计划为我们的所有数据管道场景实施 apache 气流。为此探索无法为每个任务实例/Dag 找到 unique_id 的功能。当我搜索大多数解决方案时，大多数解决方案都以宏和模板结尾。但是它们都没有为任务提供唯一 ID。但是我能够在 UI 中看到每个任务的增量唯一 ID。有什么方法可以轻松访问我的 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 会有多个依赖任务。 附加来自气流 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 而不是数字。