왜 올바른 파이프라인 구성이 중요한가요?과 Unity Catalog Co

CREATE CONNECTION mysql_prod
TYPE MYSQL
OPTIONS (
    host = 'mysql-prod.example.com',    -- MySQL 서버 주소
    port = '3306',                       -- 포트 (기본값: 3306)
    user = 'databricks_reader',          -- 접속 계정
    password = secret('scope', 'mysql-password'),  -- Secret 참조
    useSSL = 'true',                     -- SSL 암호화 사용
    trustServerCertificate = 'false'     -- 서버 인증서 검증
);

CREATE CONNECTION postgres_analytics
TYPE POSTGRESQL
OPTIONS (
    host = 'postgres.example.com',
    port = '5432',
    user = 'databricks_cdc',
    password = secret('scope', 'pg-password'),
    database = 'analytics',              -- 대상 데이터베이스 지정
    sslmode = 'require'                  -- SSL 모드 (require/verify-full)
);

CREATE CONNECTION sqlserver_erp
TYPE SQLSERVER
OPTIONS (
    host = 'sqlserver.example.com',
    port = '1433',
    user = 'databricks_reader',
    password = secret('scope', 'mssql-password'),
    database = 'ERP',
    encrypt = 'true',
    trustServerCertificate = 'false'
);

CREATE CONNECTION salesforce_crm
TYPE SALESFORCE
OPTIONS (
    sfUrl = 'https://mycompany.my.salesforce.com',
    oauthClientId = secret('scope', 'sf-client-id'),
    oauthClientSecret = secret('scope', 'sf-client-secret'),
    oauthRefreshToken = secret('scope', 'sf-refresh-token')
);

# 방법 1: 전체 스키마 수집
@dlt.table
def ingest_all():
    return mysql.read_changefeed(
        connection="mysql_prod",
        source_schema="ecommerce",
        # source_table 생략 시 전체 스키마 수집
    )

# 방법 2: 특정 테이블만 선택
@dlt.table
def ingest_customers():
    return mysql.read_changefeed(
        connection="mysql_prod",
        source_schema="ecommerce",
        source_table="customers"
    )