气流 - 更改单个任务颜色答案

【问题标题】：Airflow - Change single task color气流 - 更改单个任务颜色
【发布时间】：2020-10-02 13:40:03
【问题描述】：

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

提前致谢！

【问题讨论】：

标签： python google-cloud-platform airflow

【解决方案1】：

目前我们只能设置操作符颜色（docs）。操作员的所有任务实例都将具有相同的颜色。由于使用类变量（例如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)))

【讨论】：