Docker compose with Airflow-MS SQL Server（连接失败）

Question

我无法使用docker compose在SQL Server内部连接。我想将数据从SQL Server直接传送到云存储，然后将数据发送到Big Query。

如何解决这个问题？

import json
from datetime import timedelta, datetime

from airflow import DAG
from airflow.models import Variable
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from airflow.contrib.operators.bigquery_check_operator import BigQueryCheckOperator
from airflow.contrib.operators.file_to_gcs import FileToGoogleCloudStorageOperator
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
from airflow.contrib.operators.mysql_to_gcs import MySqlToGoogleCloudStorageOperator


default_args = {
    'owner': 'Test Data',
    'depends_on_past': True,    
    'start_date': datetime(2019, 5, 29),
    'end_date': datetime(2019, 5, 30),
    'email': ['[e-mail]'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Set Schedule: Run pipeline once a day. 
# Use cron to define exact time. Eg. 8:15am would be "15 08 * * *"
schedule_interval = "* * * * *"

# Define DAG: Set ID and assign default args and schedule interval
dag = DAG(
    'bigquery_github_trends', 
    default_args=default_args, 
    schedule_interval=schedule_interval
    )


extract = MySqlToGoogleCloudStorageOperator(
    task_id='chama_extract',
    mysql_conn_id='mysql_hml',
    google_cloud_storage_conn_id='my_gcp_conn',
    sql="""SELECT * FROM test""",
    bucket='my_bucket',
    filename='test/test{}.json',
    schema_filename='schemas/test.json',
    dag=dag)

load = GoogleCloudStorageToBigQueryOperator(
            task_id='chama_load',
            bigquery_conn_id='my_gcp_conn',
            google_cloud_storage_conn_id='my_gcp_conn',
            bucket='my_bucket',
            destination_project_dataset_table="tst.teste123",
            source_objects=['test/test0.json'],
            schema_object='schemas/test.json',
            source_format='NEWLINE_DELIMITED_JSON',
            create_disposition='CREATE_IF_NEEDED',
            write_disposition='WRITE_TRUNCATE',
            dag=dag)


# Setting up Dependencies
load.set_upstream(extract)

Docker撰写。yml

version: '3'
services:
  postgres:
    image: postgres:9.6
    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow
    ports:
      - "5432:5432"

  webserver:
    image: puckel/docker-airflow:1.10.1
    build:
      context: https://github.com/puckel/docker-airflow.git#1.10.1
      dockerfile: Dockerfile
      args:
        AIRFLOW_DEPS: gcp_api,s3        
    restart: always
    depends_on:
      - postgres
    environment:
      - LOAD_EX=n
      - EXECUTOR=Local
      - FERNET_KEY=jsDPRErfv8Z_eVTnGfF8ywd19j4pyqE3NpdUBA_oRTo=
    volumes:
      - ./examples/intro-example/dags:/usr/local/airflow/dags
      # Uncomment to include custom plugins
      # - ./plugins:/usr/local/airflow/plugins
    ports:
      - "8080:8080"
    command: webserver
    healthcheck:
      test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
      interval: 30s
      timeout: 30s
      retries: 3

docker-compose-gcloud.yml

version: '3'
services:
  postgres:
    image: postgres:9.6
    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow
    ports:
      - "5432:5432"

  webserver:
    image: puckel/docker-airflow:1.10.1
    build:
      context: https://github.com/puckel/docker-airflow.git#1.10.1
      dockerfile: Dockerfile
      args:
        AIRFLOW_DEPS: gcp_api,s3        
    restart: always
    depends_on:
      - postgres
    environment:
      - LOAD_EX=n
      - EXECUTOR=Local
      - FERNET_KEY=jsDPRErfv8Z_eVTnGfF8ywd19j4pyqE3NpdUBA_oRTo=
    volumes:
      - ./examples/gcloud-example/dags:/usr/local/airflow/dags
      # Uncomment to include custom plugins
      # - ./plugins:/usr/local/airflow/plugins
    ports:
      - "8080:8080"
    command: webserver
    healthcheck:
      test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
      interval: 30s
      timeout: 30s
      retries: 3

并在docker中执行命令：

docker compose-f docker compose gcloud。yml up—容器退出时中止

气流中的错误消息：

[2019-05-29 07:00:37938]{{logging_mixin.py:95}}信息-[2019-05-29 07:00:37937]{base_hook.py:83}信息-使用到：10.0的连接。0.1[2019-05-29 07:00:58974]{models.py:1760}}错误-（2003，“无法连接到10.0.0.1上的MySQL服务器（111“连接被拒绝”）
回溯（最近一次调用最后一次）：
文件/usr/local/lib/python3.6/site packages/afflow/models.py”，第1659行，在“运行”原始任务中。执行（context=context）
文件/usr/local/lib/python3.6/site packages/afflow/contrib/operators/mysql_to_gcs.py”，第105行，执行
cursor=self_query_mysql（）
文件“/usr/local/lib/python3.6/site packages/afflow/contrib/operators/mysql_to_gcs.py”，第127行，在query_mysql
conn=mysql中。get_conn（）
文件/usr/local/lib/python3.6/site packages/afflow/hooks/mysql_hook.py”，第103行，在get_conn
conn=MySQLdb中。connect（**conn_config）
文件/usr/local/lib/python3.6/site packages/MySQLdb/init.py”，第84行，在connect
返回连接（*args，**kwargs）
文件/usr/local/lib/python3.6/site packages/MySQLdb/connections.py）中，第164行，在init
超级（连接，self）中。init（*args，**kwargs2）
MySQLdb_例外情况。操作错误：（2003，“无法连接到10.0.0.1上的MySQL服务器（111“连接被拒绝”））
[2019-05-29 07:00:58988]{{models.py:1789}}}INFO-所有重试均失败；将任务标记为失败
[2019-05-29 07:00:58992]{{logging_mixin.py:95}}INFO-[2019-05-29 07:00:58991]{{configuration.py:255}}警告-section/key[smtp/smtp_user]在配置文件中找不到
[2019-05-29 07:00:58998]{models.py:1796}错误-[Errno 99]无法分配请求的地址
回溯（最近一次调用）：
文件“/usr/local/lib/python3。6/现场包/气流/模型。py”，第1659行，在_run_raw_task
result=task\u copy.execute（context=context）
文件/usr/local/lib/python3中。6/站点包/airflow/contrib/operators/mysql\u to\u gcs。py”，第105行，在execute
cursor=self.\u query\u mysql（）
文件/usr/local/lib/python3中。6/站点包/airflow/contrib/operators/mysql\u to\u gcs。py“，第127行，在_query_mysql
conn=mysql.get_conn（）
File”/usr/local/lib/python3中。6/站点包/气流/挂钩/挂钩。py”，第103行，在get_conn
conn=MySQLdb.connect（**conn_config）
文件/usr/local/lib/python3中。6/站点包/MySQLdb/init。py“，第84行，在Connect
返回连接（*args，**kwargs）
文件/usr/local/lib/python3中。6/站点包/MySQLdb/connections。py“，第164行，在init
super（Connection，self）.init（*args，**kwargs2）
MySQLdb._exceptions.OperationalError:（2003，“无法连接到10.0.0.1上的MySQL服务器（111“连接被拒绝”））

Answer 1

从错误来看，对我来说最关键的部分似乎是“get_conn”部分。这表示当airflow尝试建立与数据库的连接时失败。这意味着您的连接未指定（看起来可能是），或者部分连接不正确。

您应该检查密码、服务器地址和端口是否正确。这些应该是在你的气流。cfg，作为环境变量，或在Web服务器（管理面板）中设置