Airflow - 更改单个任务颜色

Question

有人知道是否可以更改 Airflow UI 中的单个任务颜色吗？我读到可以更改运算符颜色（导入它后，您可以覆盖类方法），但是如果我有两个具有不同颜色的 PythonOperator 实例怎么办？

提前致谢！

Answer 1

目前我们只能设置操作符颜色（文档）。操作员的所有任务实例都将具有相同的颜色。由于使用类变量（例如PythonOperator.ui_color，PythonOperatpr.ui_fgcolor）来设置颜色，因此不可能根据任务实例更改颜色。

一种方法是继承PythonOperator每种颜色，如下所示：

class BluePythonOperator(PythonOperator):
    ui_color = "blue"
    ui_fgcolor = "white"


class BlackPythonOperator(PythonOperator):
    ui_color = "black"
    ui_fgcolor = "white"


with DAG(dag_id='my_example',
         default_args=default_args) as dag0:
    t0 = BluePythonOperator(
        task_id="blue_colored_task",
        python_callable=(lambda x: print(x)))
    t1 = BlackPythonOperator(
        task_id="black_colored_task",
        python_callable=(lambda x: print(x)))

但是，如果有更多颜色，我们将不得不创建许多继承类。

所以我尝试创建一个类工厂之类的东西，它返回一个子类，PythonOperator并基于task_id.

def colorful_python_operator(task_id):
    tid_to_colors = {
        "blue_colored_task": {"ui_color": "blue", "ui_fgcolor": "white"},
        "black_colored_task": {"ui_color": "black", "ui_fgcolor": "white"}
    }

    class ColorfulPythonOperator(PythonOperator):
        value = tid_to_colors[task_id]
        ui_color = value["ui_color"]
        ui_fgcolor = value["ui_fgcolor"]
        pass

    return ColorfulPythonOperator


with DAG(dag_id='my_another_example',
         default_args=default_args) as dag1:
    t2 = colorful_python_operator("blue_colored_task")(
        task_id="blue_colored_task",
        python_callable=(lambda x: print(x)))

    t3 = colorful_python_operator("black_colored_task")(
        task_id="black_colored_task",
        python_callable=(lambda x: print(x)))