kubernetes - 如何在GCP Cloud Composer上的Apache Airflow上将带有 `KubernetesPodOperator`的连接 Hook 用作环境变量

Question

我想在使用airflow的任务中使用connections saved in KubernetesPodOperator 。

在开发镜像时，我使用环境变量将数据库连接信息向下传递到容器，但是生产环境将数据库另存为连接 Hook 。

提取数据库连接信息并将其传递给容器的最佳方法是什么？

env_vars = {'database_usr': 'xxx', 'database_pas': 'xxx'}

KubernetesPodOperator(
        dag=dag,
        task_id="example-task",
        name="example-task",
        namespace="default",
        image="eu.gcr.io/repo/image:tag",
        image_pull_policy="Always",
        arguments=["-v", "image-command", "image-arg"],
        env_vars=env_vars,
    )

Answer 1

我当前的解决方案是使用BaseHook从连接中获取变量:

from airflow.hooks.base_hook import BaseHook


def connection_to_dict(connection_id):
    """Returns connection params from Airflow as a dictionary.

    Parameters
    ----------
    connection_id : str
        Name of the connection in Airflow, e.g. `mysql_default`

    Returns
    -------
    dict
        Unencrypted values.
    """
    conn_obj = BaseHook.get_connection(connection_id)
    d = conn_obj.__dict__
    if ('is_encrypted', True) in d.items():
        d['password'] = conn_obj.get_password()
    return d

然后将它们作为环境变量传递给Kubernetes Pod运算符。