Airflow 中的动态用户模拟

如何解决Airflow 中的动态用户模拟

是否有一种简单有效的机制可以根据触发 DAG 执行的用户为 Airflow 2.0 任务实例动态设置 run_as_user 参数？

This answer 提供了一种通过检查元数据数据库来确定触发用户的方法（涉及模板的早期答案——例如 this——似乎不再适用于 Airflow 2.0）。使用元数据检查方法，我已经能够通过客户运算符动态设置 run_as_user，如下所示：

from airflow.operators.python import Pythonoperator
from airflow.models.log import Log
from airflow.utils.db import create_session

class CustomPythonoperator(Pythonoperator):
    def __init__(self,**kwargs):
        super().__init__(**kwargs)
 
        with create_session() as session:
            results = session.query(Log.owner)\
                .filter(
                    Log.dag_id==dag.dag_id,Log.event=='trigger')\
                .order_by(Log.dttm.desc())\
                .all()
            
        self.run_as_user = results[0][0] if len(results) else 'airflow'

也就是说，这种方法似乎有问题，原因至少有两个：

1. 表面上，它要求调度程序在每次实例化自定义操作符时（而不是每次执行任务时）查询元数据数据库。 documentation 警告自定义运算符的 __init__ 方法中的昂贵操作。将用户查找移动到自定义运算符的 execute 方法不起作用。
2. 我怀疑这个解决方案的语义不完全：具体来说，我怀疑这种方法会将 run_as_user 设置为最近触发 DAG 的用户调度程序解析 DAG 的时间，而不是与当前 DAG 执行关联的用户。根据调度程序间隔，他们可能是同一个用户，但这似乎会引入竞争条件。

有没有更好的方法来解决这个问题？