代码之家 › 专栏 › 技术社区 › Kyle Bridenstine

获取重试号码

airflow

1

Kyle Bridenstine · 技术社区 · 7 年前

在我的Airflow DAG中,我有一个任务需要知道它是第一次运行还是重试运行。如果是重试尝试,我需要调整任务中的逻辑。

关于如何存储任务的重试次数,我有一些想法,但我不确定其中是否有合法的,或者是否有更简单的内置方法来获取任务中的信息。

我想知道我是否可以在dag中有一个整数变量,每次任务运行时都会附加这个变量。然后,如果任务if重新运行,我可以检查变量的值,看看它是否大于1,因此将是一次重试运行。但我不确定可变全局变量是否在气流中以这种方式工作,因为可能有多个工作人员负责不同的任务(但我不确定这一点)。
将其写入XCOM变量?

1 回复 | 直到 7 年前

1

16

cwurtz 7 年前

重试次数可从任务实例获得,该任务实例可通过宏获得 {{ task_instance }} . https://airflow.apache.org/code.html#default-variables

如果您使用的是python操作符,只需添加 provide_context=True, 给你的接线员kwargs,然后打电话 kwargs['task_instance'].try_number

否则,您可以执行以下操作:

t = BashOperator(
    task_id='try_number_test',
    bash_command='echo "{{ task_instance.try_number }}"',
    dag=dag)

编辑:

当任务实例被清除时,它会将max_retry number设置为当前try_number+retry值。所以你可以这样做:

ti = # whatever method you do to get the task_instance object
is_first = ti.max_tries - ti.task.retries + 1 == ti.try_number

在运行时,Airflow会将try_数增加1,所以我想在从配置的重试值中减去max_trytes时,您需要+1。但我并不是为了证实这一点而测试的

2

3

user3505886 4 年前

@科沃茨的回答恰到好处。我可以这样使用它:

    def _get_actual_try_number(self, context):
        '''
        Returns the real try_number that you also see in task details or logs.
        '''
        return context['task_instance'].try_number

    def _get_relative_try_number(self, context):
        '''
        When a task is cleared, the try_numbers continue to increment.
        This returns the try number relative to the last clearing.
        '''
        ti = context['task_instance']
        actual_try_number = self._get_actual_try_number(context)

        # When the task instance is cleared, it will set the max_retry
        # number to be the current try_number + retry value.
        # From https://stackoverflow.com/a/51757521
        relative_first_try = ti.max_tries - ti.task.retries + 1

        return actual_try_number - relative_first_try + 1