如何在 Airflow 的函数中使用宏答案

【问题标题】：How to use macros within functions in Airflow如何在 Airflow 的函数中使用宏
【发布时间】：2019-10-01 17:11:49
【问题描述】：

我正在尝试使用 dag_id、task_id 和 execution_date 的组合来计算气流中每个任务的哈希值。我正在自定义运算符的 init 中进行计算，以便我可以使用它来计算每个任务的唯一 retry_delay（我不想使用指数退避）

我发现很难在调用散列函数或 int 函数中使用 {{ execution_date}} 宏，在这些情况下，气流不会将其替换为特定日期（只保留字符串 {{execution_date}} 我得到所有执行日期都一样

self.task_hash = int(hashlib.sha1("{}#{}#{}".format(self.dag_id,
                                                    self.task_id,
                                                    '{{execution_date}}')
                                       .encode('utf-8')).hexdigest(), 16)

我已将 task_hash 放入 template_fields，我也尝试在自定义宏中进行计算 - 这适用于哈希部分，但当我将其放入 int() 时，它是同样的问题

任何解决方法，或者我可以检索 execution_date（在运算符的 init 上），而不是从宏中？

谢谢

【问题讨论】：

标签： airflow

【解决方案1】：

试试：

self.task_hash = int(hashlib.sha1("{}#{}#{{execution_date}}".format(
    self.dag_id, self.task_id).encode('utf-8')).hexdigest(), 16)

【讨论】：