Это просто ! Нет необходимости в DockerFile, KubernetesPodOperator, LD_LIBRARY_PATH и т. Д. Достаточно простого оператора python.
Что следует учитывать
- Образ GCP Composer Worker's Pod - это ubuntu 1604 (для проверки просто запустите базовый оператор Python с помощью команды os.system ('cat / etc / os-release'))
- У него уже установлен unixodbc-dev в образе пода рабочего.
- Композитор создает ковш и устанавливает его с помощью воздушного потока
- Так почему бы просто не установить pyodbc из пакетов pypi и предоставить драйвер mssql odbc в качестве параметра в методе подключения pyodbc
здесь 'gs: // bucket_created_by_composer' == '/ home / airflow / gcs'
gcs bucket created by composer ->
-> data/
-> dags/
Пошаговый подход
Шаг 1. Установите pyodbc, mssql odbc на все экземпляры ubuntu, чтобы получить файлы драйверов.
для рассмотрения, давайте сделаем это на GCP VM Intance с изображением ubuntu 1804
#update the packages
sudo apt update
sudo apt-get update -y
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/18.04/prod.list | sudo tee /etc/apt/sources.list.d/msprod.list
sudo apt-get update -y
echo Installing mssql-tools and unixODBC developer...
sudo ACCEPT_EULA=Y apt-get install -y mssql-tools unixodbc-dev
sudo apt-get update -y
sudo apt-get install -y mssql-tools #it includes sql_cmd and bcp (we dont need those)
sudo apt install python3-pip #installing pip3
pip3 install pyodbc
Шаг 2: получите файлы драйвера и загрузите их в папку данных gcs_bucket, созданную композитором.
cd /opt/microsoft
#now you can see there is one directory 'msodbcsql17', version may change
#we need to upload this directory to the data folder of gcs_bucket
#for this you may choose which ever approach suits you
#copying the directory to /<home/user> for proper zipping/uploading to gcs
cp -r msodbcsql17 /home/<user> #you may need to use sudo
#upload this /home/<user>/msodbcsql17 to any gcs_bucket
gsutil cp -r /home/<user>/msodbcsql17 gs://<your-gcs-bucket>
загрузите эту папку из ведра gcs на локальный компьютер и загрузите эту папку в папку данных ведра gcs, созданного композитором
выберите любой подход / метод, основная цель - получить папку msodbcsql17 в папке данных ведра gcs, созданного композитором
Окончательная структура:
gcs bucket created by composer ->
-> data/msodbcsql17/
-> dags/<your_dags.py>
Шаг 3: использование драйверов msodbcsql17 для подключения pyodbc
ПРИМЕР DAG:
import os
import time
import datetime
import argparse
import json
from airflow import DAG
import airflow
from airflow.operators import python_operator
default_dag_args = {
'start_date': airflow.utils.dates.days_ago(0), #
'provide_context': True
}
dag = DAG(
'pyodbc_test',
schedule_interval=None, #change for composer
default_args=default_dag_args
)
def check_connection(**kwargs):
print('hello')
driver='/home/airflow/gcs/data/msodbcsql17/lib64/libmsodbcsql-17.5.so.2.1'
#this is the main driver file, the exact location can be found on gcs_bucket/data folder or check the /etc/odbcinst.in file of ubuntu instance in which you installed the pyodbc earlier
def tconnection(ServerIp,LoginName,Password,mssql_portno):
""" A method which return connection object"""
import pyodbc
pyodbc.pooling = False
try:
sql_conn = pyodbc.connect("DRIVER={4};SERVER={0},{1};UID={2};PWD={3}".format(ServerIp,mssql_portno,LoginName,Password,driver))
except pyodbc.Error as ex:
sqlstate = ex.args[1]
raise
return sql_conn
con=tconnection('<your-server-ip>','<your-login-name>','<your-password>','1433')
#recommendation is to take the password and login from airflow connections
import pandas as pd
q='select * from <your-db-name>.<your-schema-name>.<your-table-name>'
df=pd.read_sql(q,con)
print(df)
Tcheck_connection= python_operator.PythonOperator(
task_id='Tcheck_connection',
python_callable=check_connection,
dag=dag )
#calling the task sequence
Tcheck_connection
Пакеты PYPI
pyodbc
pandas
Недавно протестировали на Composer
person
D.C
schedule
30.06.2020
