Microsoft SQL Server 연결과 CDC

Microsoft SQL Server 커넥터는 SQL Server 데이터베이스와 Foundry 사이의 원천 탐색, 배치 동기화, 증분 수집, Iceberg 동기화, CDC와 테이블 내보내기를 일반 제공(GA)합니다. Data Connection의 +New Source에서는 MS SQL Server 소스를 선택합니다.

인증 방식

Azure Active Directory의 현재 이름은 Microsoft Entra ID이지만 JDBC 옵션은 기존 Active Directory 이름을 유지합니다. 모든 방식에서 사용자와 역할에 데이터베이스 접근 및 대상 테이블의 읽기 또는 쓰기 권한이 필요합니다.

네트워크와 Azure 연결 정책

일반적인 연결은 호스트 이름과 포트 1433을 사용합니다. 소스에 대상 외부 통신 정책을 추가하며 Azure SQL이나 AWS RDS에서는 클라우드 콘솔에서 인스턴스의 실제 호스트 이름을 확인합니다.

Azure 포털의 Settings, Properties에서 Server Name을 확인할 수 있습니다. 호스트 기반 정책의 형식은 <your-database-name>.database.windows.net의 1433 포트입니다. nslookup으로 해석한 IP를 확인할 수 있고 부하 분산 때문에 여러 IP가 반환될 수 있습니다.

Azure Redirect 모드를 사용하면 해석된 모든 IP에 대한 외부 통신 정책도 필요합니다. IP가 바뀌면 허용 목록을 갱신해야 합니다.

Foundry와 Azure SQL이 모두 Azure에 있으면 Azure SQL 연결 정책의 Proxy를 사용합니다. Azure 내부 트래픽의 기본 정책인 Redirect는 Azure SQL의 모든 IP에 대해 11000~11999 포트를 허용해야 할 수 있어 지나치게 넓은 구성이 됩니다. 여기서 Proxy는 Azure SQL의 연결 정책이며 Foundry의 에이전트 프록시 설정과 구분합니다.

연결 속성

CDC 활성화

데이터베이스와 각 대상 테이블에 CDC를 활성화합니다. 다음 자리표시자는 실제 데이터베이스, 스키마와 테이블 이름으로 바꿉니다. 명령은 원천 데이터베이스 설정을 변경합니다.

USE <database>
GO
EXEC sys.sp_cdc_enable_db
GO

EXEC sys.sp_cdc_enable_table
    @source_schema = N'<schema>'
  , @source_name = N'<table_name>'
  , @role_name = NULL
  , @capture_instance = NULL
  , @supports_net_changes = 0
  , @filegroup_name = N'PRIMARY';
GO

원천 탐색 실행 환경이 정상 동작한 뒤 Overview의 + Create CDC sync를 선택합니다. 초기화 중이면 몇 초 기다리고 페이지를 새로고침합니다.

CDC 읽기 권한

CDC 스키마의 실행 권한은 다음 질의로 확인합니다. 1이면 권한이 있고 0이면 없습니다.

SELECT HAS_PERMS_BY_NAME('cdc', 'SCHEMA', 'EXECUTE') AS HasExecutePermission;

권한이 부족하면 원천 시스템에서 실제 대상 사용자에 권한을 부여합니다.

GRANT EXECUTE ON SCHEMA::cdc TO <USER>;

Active Directory Password 로그인 오류

Login failed for user ''. 오류는 서버가 Kerberos 또는 NTLM만 허용하고 Azure Active Directory 인증을 받지 않는 경우에 나타날 수 있습니다.

ActiveDirectoryPassword는 Azure Active Directory에 인증하며 Kerberos나 NTLM 방식이 아닙니다. 이 경우 사용자 정의 JDBC 연결에서 authenticationScheme=NTLM을 설정하여 NTLM 경로를 사용합니다.

외부 변환의 예시 테이블

아래 Fruits는 예시 테이블입니다. 테이블 생성은 데이터베이스를 변경하므로 비운영 대상을 사용합니다.

CREATE TABLE Fruits (
    FruitName VARCHAR(50) PRIMARY KEY,
    Inventory INT NOT NULL,
    PricePerKg DECIMAL(10, 2) NOT NULL
);

Python 저장소에 pymssql과 pandas를 추가하고 소스를 가져옵니다. 코드의 소스 RID, 데이터셋 RID, 서버, 데이터베이스와 사용자 이름을 실제 구성으로 바꿉니다. 소스에는 MSSQL_BASIC_AUTH_PASSWORD 비밀 정보가 있어야 합니다.

재고가 적은 행 읽기

Inventory < 60인 행을 읽어 pandas 결과를 반환합니다. 값은 SQL 매개변수로 전달하며 연결은 암호화를 요구하고 timeout을 30초로 지정합니다. 오류를 로그에 남기고 RuntimeError로 전달합니다.

import logging
from pandas import DataFrame
from transforms.external.systems import ResolvedSource
from transforms.api import lightweight, Output, transform_pandas
from transforms.external.systems import external_systems, Source
import pymssql
import pandas as pd

logger = logging.getLogger(__name__)

@lightweight
@external_systems(
    mssql_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<dataset_rid>")
)
def compute(mssql_source: ResolvedSource) -> DataFrame:
    # Inventory threshold parameter (this could also be read from an input DataFrame)
    INVENTORY_THRESHOLD = 60

    connection_parameters = {
        "server": "<your_server_name>",
        "database": "<your_database_name>",
        "port": "1433",
        "user": "<your_user_name>",
        "password": mssql_source.get_secret("MSSQL_BASIC_AUTH_PASSWORD"),
        "encryption": "require",
        "timeout": 30
    }
    try:
        with pymssql.connect(**connection_parameters) as connection:
            df = pd.read_sql(
                'SELECT * FROM Fruits WHERE Inventory < %s',
                connection,
                params=(INVENTORY_THRESHOLD,)
            )
    except Exception as e:
        logger.error(f"Error querying MSSQL Fruits table: {e}")
        raise RuntimeError(f"Failed to fetch Fruits data from MSSQL: {e}") from e
    return df

행 갱신 또는 삽입

입력 데이터셋의 열은 FruitName, Inventory, PricePerKg입니다. FruitName으로 먼저 UPDATE를 수행하고 갱신한 행이 없으면 INSERT합니다. 작업을 완료한 뒤 커밋하고 과일 이름별 updated 또는 inserted 결과를 반환합니다.

입력 데이터셋과 작업 결과 출력은 서로 다른 실제 데이터셋으로 지정합니다. 실행하면 외부 테이블이 변경되므로 대상과 입력을 확인합니다.

import logging
from transforms.api import lightweight, Output, transform_pandas, Input
from transforms.external.systems import external_systems, Source
import pymssql
import pandas as pd

logger = logging.getLogger(__name__)

@lightweight
@external_systems(
    mssql_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<output_dataset_rid>"),
    fruits_df=Input("<input_dataset_rid>") # DataFrame with schema [FruitName: String, Inventory: Integer, PricePerKg: Double]
)
def compute(mssql_source, fruits_df):
    # Connection parameters
    connection_parameters = {
        "server": "<your_server_name>",
        "database": "<your_database_name>",
        "port": "1433",
        "user": "<your_user_name>",
        "password": mssql_source.get_secret("MSSQL_BASIC_AUTH_PASSWORD"),
        "encryption": "require",
        "timeout": 30
    }
    results = []
    try:
        with pymssql.connect(**connection_parameters) as connection:
            with connection.cursor() as cursor:
                for _, row in fruits_df.iterrows():
                    fruit_name = row["FruitName"]
                    inventory = row["Inventory"]
                    price_per_kg = row["PricePerKg"]

                    # Try to update; if no row updated, then insert
                    update_sql = """
                        UPDATE Fruits
                        SET Inventory = %s, PricePerKg = %s
                        WHERE FruitName = %s
                    """
                    cursor.execute(update_sql, (inventory, price_per_kg, fruit_name))
                    if cursor.rowcount == 0:
                        insert_sql = """
                            INSERT INTO Fruits (FruitName, Inventory, PricePerKg)
                            VALUES (%s, %s, %s)
                        """
                        cursor.execute(insert_sql, (fruit_name, inventory, price_per_kg))
                        results.append({"FruitName": fruit_name, "action": "inserted"})
                    else:
                        results.append({"FruitName": fruit_name, "action": "updated"})
                connection.commit()
    except Exception as e:
        logger.error(f"Error updating MSSQL Fruits table: {e}")
        raise RuntimeError(f"Failed to update Fruits data in MSSQL: {e}") from e
    # Return a DataFrame summarizing the actions
    return pd.DataFrame(results, columns=["FruitName", "action"])