Google和

Question

我们有什么方法可以从新引入的连接到Google (MySQL)实例？这样做的目的是将数据从Cloud实例中获取到BigQuery中(也许是通过Cloud的中间步骤)。

1. Cloud代理能否以某种方式公开在Kubernetes集群宿主编写器中的pods部分？
2. 如果没有，可以使用Kubernetes Service？-> https://cloud.google.com/kubernetes-engine/docs/concepts/add-on/service-broker引入Cloud代理。
3. 是否应该使用气流来调度和调用GCP命令，例如: 1)将mysql表导出到云存储，2)将mysql导出到bigquery？
4. 也许我还缺少其他方法来完成这个任务。

Answer 1

“Cloud代理提供对Cloud第二代实例的安全访问，而无需白名单IP地址或配置SSL。”-代理文档。

CloudSQL代理似乎是最值得推荐的连接到CloudSQL的方法。因此，在Composer中，从1.6.1版开始，我们可以创建一个新的Kubernetes Pod来运行gcr.io/cloudsql-docker/gce-代理:最新的映像，通过服务公开它，然后在Composer中创建一个连接，以便在操作符中使用。

要建立起来：

• 遵循Google文档
• 使用阿里克(氏)中型邮政中的信息测试连接

- Check that the pod was created `kubectl get pods --all-namespaces`
- Check that the service was created `kubectl get services --all-namespaces`
- Jump into a worker node `kubectl --namespace=composer-1-6-1-airflow-1-10-1-<some-uid> exec -it airflow-worker-<some-uid> bash`

    - Test mysql connection `mysql -u composer -p --host <service-name>.default.svc.cluster.local`

​

​

备注：

• Composer现在使用命名空间来组织荚
• 不同名称空间中的不要和对方说话，除非您给他们提供完整的路径<k8-service-name>.<k8-namespace-name>.svc.cluster.local
• 创建具有完整路径的新作曲家连接将启用成功的连接。

Answer 2

我们也遇到了同样的问题，但对于Postgres实例。这就是我们所做的，并让它发挥作用：

• 在运行气流的Kubernetes集群中创建sqlproxy部署。这是默认airflow_db连接使用的现有气流-sqlproxy的副本，对部署文件进行了以下更改：

- replace all instances of airflow-sqlproxy with the new proxy name
- edit under 'spec: template: spec: containers: command: -instances', replace the existing instance name with the new instance we want to connect to

​

• 创建一个kubernetes服务，同样作为现有气流-sqlproxy服务的副本，并进行以下更改：

- replace all instances of airflow-sqlproxy with the new proxy name
- under 'spec: ports', change to the appropriate port (we used 5432 for a Postgres instance)

​

• 在气流UI中，将具有主机设置的Postgres类型的连接添加到新创建的服务名称中。

Answer 3

在从@Leo到顶级https://medium.com/@ariklevliber/connecting-to-gcp-composer-tasks-to-cloud-sql-7566350c5f53的注释中添加中间帖子。完成这篇文章并完成服务设置之后，您可以使用SQLAlchemy从DAG连接，如下所示：

import os
from datetime import datetime, timedelta
import logging

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator

logger = logging.getLogger(os.path.basename(__file__))
INSTANCE_CONNECTION_NAME = "phil-new:us-east1:phil-db"

default_args = {
    'start_date': datetime(2019, 7, 16)
}


def connect_to_cloud_sql():
    '''
        Create a connection to CloudSQL
    :return:
    '''
    import sqlalchemy
    try:
        PROXY_DB_URL = "mysql+pymysql://<user>:<password>@<cluster_ip>:3306/<dbname>"
        logger.info("DB URL", PROXY_DB_URL)
        engine = sqlalchemy.create_engine(PROXY_DB_URL, echo=True)
        for result in engine.execute("SELECT NOW() as now"):
            logger.info(dict(result))
    except Exception:
        logger.exception("Unable to interact with CloudSQL")


dag = DAG(
    dag_id="example_sqlalchemy",
    default_args=default_args,
    # schedule_interval=timedelta(minutes=5),
    catchup=False  # If you don't set this then the dag will run according to start date
)


t1 = PythonOperator(
    task_id="example_sqlalchemy",
    python_callable=connect_to_cloud_sql,
    dag=dag
)


if __name__ == "__main__":
    connect_to_cloud_sql()