Airflow - KubernetesPodOperator 中的条件重试

Question

我想在我的 kube pod operator 中编写高级重试机制。

1. 如果第一次执行失败 - 使用更多内存重新运行 上面带有 jinja 模板的示例不起作用:-(

opr_kube = KubernetesPodOperator(
    namespace="development",
    image='bla',
    node_selectors={"servertype": "worker", "env": "development"},
    resources=pod.Resources(request_memory='{{task_instance.try_number}}g', limit_memory='{{task_instance.try_number}}g'),
    name="bla",
    task_id='bla',
    default_args=default_args,
    dag=dag
)

1. 我想了解错误是否是由于内存不足事件造成的。有一个界面可以识别吗？我想写外部函数到 kube-cli

   • on-retry 钩子在这里不相关，因为没有选项可以用更多内存再次触发运算符。
   • 当我在参数内部添加内联函数时，调度程序将每 X 秒执行一次此函数，我想在任务失败时执行它。

Answer 1

我想了解错误是否是因为内存不足 事件：

关于任务实例失败的原因，请查看 Airflow 网页界面 => DAG 的 Graph View

关于 Kubernetes Operator 重试选项，here 是一个示例，但您应该首先了解任务失败的原因。如果不验证问题的真正原因，就没有理由使用增量内存重试机制。请事先确保您通过 KubernetesPodOperator 运行的工作负载有足够的资源（您也可以参考“Ensuring appropriate resources for your environment”指南，该指南更专门针对 Airflow 的托管版本）