Airflow에서 15년 된 SQL Server 연결하기: pymssql 실패부터 Custom Hook까지

DB-Lib error message 20002, severity 9:
Adaptive Server connection failed

pymssql (FreeTDS)
└─ 프로토콜 핸드셰이크 실패
   └─ DB-Lib error 20002

jTDS (DBeaver)
└─ 레거시 TDS 프로토콜까지 유연하게 지원
   └─ SQL Server 2008 R2와 호환 → 연결 성공

Python 코드 (Airflow Task)
    ↓
DB-API 2.0 (PEP 249)
    ↓
pyodbc (DB-API 구현체)
    ↓
ODBC Driver 18 for SQL Server
    ↓
TDS 프로토콜
    ↓
SQL Server 2008 R2

FROM apache/airflow:3.1.0

USER root

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        curl gnupg2 ca-certificates ffmpeg \
        unixodbc unixodbc-dev && \
    # Microsoft GPG 키 추가 (gpg --dearmor 방식)
    curl -fsSL https://packages.microsoft.com/keys/microsoft.asc | \
        gpg --dearmor -o /usr/share/keyrings/microsoft-prod.gpg && \
    echo "deb [signed-by=/usr/share/keyrings/microsoft-prod.gpg] \
        https://packages.microsoft.com/debian/12/prod bookworm main" \
        > /etc/apt/sources.list.d/mssql-release.list && \
    apt-get update && \
    ACCEPT_EULA=Y apt-get install -y msodbcsql18 && \
    # SQL Server 2008 R2 호환을 위한 OpenSSL 레거시 SSL 설정
    printf '\n[openssl_init]\nssl_conf = ssl_sect\n\n[ssl_sect]\nsystem_default = system_default_sect\n\n[system_default_sect]\nOptions = UnsafeLegacyRenegotiation\n' \
        >> /etc/ssl/openssl.cnf && \
    apt-get clean && rm -rf /var/lib/apt/lists/*

USER airflow

SSL Provider:
[error:0A000152:SSL routines::unsafe legacy renegotiation disabled]

# pyproject.toml
dependencies = [
    "apache-airflow-providers-microsoft-mssql>=4.4.0",  # UI에서 Connection 타입 선택용
    "pyodbc>=5.3.0",  # 실제 MSSQL 연결용
]

import pyodbc
from airflow.hooks.base import BaseHook


class MSSQLHook(BaseHook):
    conn_name_attr = "mssql_conn_id"
    default_conn_name = "mssql_default"

    def __init__(self, mssql_conn_id="mssql_default"):
        super().__init__()
        self.mssql_conn_id = mssql_conn_id

    def get_conn(self):
        conn = self.get_connection(self.mssql_conn_id)
        extra = conn.extra_dejson
        charset = extra.get("charset")

        connection_string = (
            f"DRIVER={{ODBC Driver 18 for SQL Server}};"
            f"SERVER={conn.host},{conn.port or 1433};"
            f"DATABASE={conn.schema};"
            f"UID={conn.login};"
            f"PWD={conn.password};"
            f"Encrypt=no;"
            f"TrustServerCertificate=yes;"
        )

        client = pyodbc.connect(connection_string)

        if charset:
            try:
                client.setdecoding(pyodbc.SQL_CHAR, encoding=charset)
                client.setdecoding(pyodbc.SQL_WCHAR, encoding=charset)
                client.setencoding(encoding=charset)
            except Exception:
                pass  # 인코딩 미지원 시 기본값 사용

        return client

    def run(self, sql):
        client = self.get_conn()
        try:
            cursor = client.cursor()
            cursor.execute(sql)
            results = cursor.fetchall()
            cursor.close()
            return results
        finally:
            client.close()

// Airflow Connection Extra (charset이 필요한 경우만)
{"charset": "cp949"}

1. ClickHouse에서 마지막 동기화된 max_idx 조회
2. MSSQL에서 max_idx 이후 데이터를 TOP N으로 배치 조회
3. pyodbc.Row → list 변환 + 데이터 검증
4. ClickHouse에 배치 INSERT
5. 다음 배치로 이동 (current_idx 갱신)
6. 전부 끝나면 OPTIMIZE TABLE FINAL (중복 제거)

@task(
    retries=1,
    retry_delay=timedelta(seconds=10),
    on_failure_callback=task_failure_callback,
)
def sync_orders(last_idx: int, batch_size: int = 1000):
    mssql_hook = MSSQLHook(mssql_conn_id="mssql_default")
    ch_hook = ClickHouseHook(clickhouse_conn_id="clickhouse")

    # MSSQL에서 최대 ID 조회
    max_result = mssql_hook.run("SELECT MAX(id) FROM orders")
    max_idx = max_result[0][0] if max_result and max_result[0][0] else 0

    if last_idx >= max_idx:
        logging.info("No new data to sync")
        return

    # 배치 단위 동기화
    ch_conn = ch_hook.get_conn()
    current_idx = last_idx
    total_synced = 0

    while current_idx < max_idx:
        sql = f"""
        SELECT TOP {batch_size} *
        FROM orders
        WHERE id > {current_idx}
        ORDER BY id
        """
        batch_data = mssql_hook.run(sql)

        if not batch_data:
            break

        # pyodbc.Row → list 변환
        processed = [list(row) for row in batch_data]

        ch_conn.insert(
            table="orders",
            data=processed,
            column_names=[...]
        )

        current_idx = processed[-1][0]
        total_synced += len(processed)
        logging.info(f"Synced {len(processed)} rows, total: {total_synced}")

    # ReplacingMergeTree 중복 제거
    try:
        ch_hook.run("OPTIMIZE TABLE orders FINAL")
    except Exception as e:
        logging.warning(f"OPTIMIZE TABLE failed (non-critical): {e}")