Lakeflow Connect 실습

-- MySQL 서버 설정 확인 (my.cnf 또는 파라미터 그룹)
-- 아래 설정이 활성화되어 있어야 합니다
SHOW VARIABLES LIKE 'log_bin';           -- ON 이어야 함
SHOW VARIABLES LIKE 'binlog_format';     -- ROW 이어야 함
SHOW VARIABLES LIKE 'binlog_row_image';  -- FULL 이어야 함

-- CDC 수집용 계정 생성 (최소 권한 원칙)
CREATE USER 'databricks_cdc'@'%' IDENTIFIED BY '<strong-password>';

-- 필수 권한 부여
GRANT SELECT ON ecommerce.* TO 'databricks_cdc'@'%';
GRANT REPLICATION SLAVE ON *.* TO 'databricks_cdc'@'%';
GRANT REPLICATION CLIENT ON *.* TO 'databricks_cdc'@'%';
FLUSH PRIVILEGES;

-- 대상 카탈로그 및 스키마 생성
CREATE CATALOG IF NOT EXISTS analytics;
CREATE SCHEMA IF NOT EXISTS analytics.bronze;
CREATE SCHEMA IF NOT EXISTS analytics.silver;

-- 파이프라인 실행 서비스 프린시펄에 권한 부여
GRANT USE CATALOG ON CATALOG analytics TO `lakeflow-connect-service`;
GRANT USE SCHEMA ON SCHEMA analytics.bronze TO `lakeflow-connect-service`;
GRANT CREATE TABLE ON SCHEMA analytics.bronze TO `lakeflow-connect-service`;

-- MySQL Connection 생성
CREATE CONNECTION mysql_ecommerce
TYPE MYSQL
OPTIONS (
    host = 'mysql-prod.company.com',
    port = '3306',
    user = 'databricks_cdc',
    password = secret('db-secrets', 'mysql-ecommerce-pwd')
);

-- 생성된 Connection 확인
SHOW CONNECTIONS;

-- Connection을 통해 소스 테이블 목록 확인
SHOW TABLES IN mysql_ecommerce.ecommerce;

import dlt
from dlt.sources import mysql

# Lakeflow Connect MySQL 수집 정의
@dlt.table(
    name="bronze_customers",
    comment="MySQL ecommerce.customers 실시간 CDC 수집"
)
def ingest_customers():
    return mysql.read_changefeed(
        connection="mysql_ecommerce",
        source_schema="ecommerce",
        source_table="customers"
    )

@dlt.table(
    name="bronze_orders",
    comment="MySQL ecommerce.orders 실시간 CDC 수집"
)
def ingest_orders():
    return mysql.read_changefeed(
        connection="mysql_ecommerce",
        source_schema="ecommerce",
        source_table="orders"
    )

-- 수집된 고객 데이터 확인
SELECT * FROM analytics.bronze.customers LIMIT 10;

-- 수집된 주문 데이터 확인
SELECT COUNT(*) AS total_orders,
       MIN(_ingested_at) AS first_ingested,
       MAX(_ingested_at) AS last_ingested
FROM analytics.bronze.orders;

-- 시스템 메타데이터 컬럼 확인
-- Lakeflow Connect가 자동으로 추가하는 메타데이터
SELECT _ingested_at,        -- 수집 시간
       _change_type,        -- INSERT/UPDATE/DELETE
       _commit_timestamp,   -- 소스 변경 시간
       customer_id, name, email
FROM analytics.bronze.customers
ORDER BY _ingested_at DESC
LIMIT 20;

-- [MySQL에서 실행] 새 고객 추가
INSERT INTO ecommerce.customers (customer_id, name, email, city)
VALUES (1001, '김테스트', 'test@example.com', 'Seoul');

-- [MySQL에서 실행] 기존 고객 정보 수정
UPDATE ecommerce.customers
SET city = 'Busan'
WHERE customer_id = 1001;

-- [MySQL에서 실행] 새 주문 추가
INSERT INTO ecommerce.orders (order_id, customer_id, amount, order_date)
VALUES (5001, 1001, 50000, NOW());

-- CDC 변경사항 확인 (수초~수분 후)
SELECT _change_type, _commit_timestamp, customer_id, name, city
FROM analytics.bronze.customers
WHERE customer_id = 1001
ORDER BY _commit_timestamp DESC;

-- 결과 예시:
-- 예상 결과:
-- | _change_type | _commit_timestamp  | customer_id | name   | city  |
-- |-------------|-------------------|-------------|--------|-------|
-- | UPDATE      | 2025-01-15 10:01  | 1001        | 김테스트 | Busan |
-- | INSERT      | 2025-01-15 10:00  | 1001        | 김테스트 | Seoul |

-- 주문 테이블 변경 확인
SELECT * FROM analytics.bronze.orders
WHERE order_id = 5001;

-- [MySQL에서 실행] 고객 테이블에 새 컬럼 추가
ALTER TABLE ecommerce.customers ADD COLUMN phone VARCHAR(20);

-- [MySQL에서 실행] 새 컬럼에 데이터 입력
UPDATE ecommerce.customers
SET phone = '010-1234-5678'
WHERE customer_id = 1001;

-- 대상 테이블에 자동으로 phone 컬럼이 추가되었는지 확인
DESCRIBE analytics.bronze.customers;

-- 새 컬럼의 데이터 확인
SELECT customer_id, name, phone
FROM analytics.bronze.customers
WHERE phone IS NOT NULL;

-- 기존 행에서 phone은 NULL (스키마 추가 이전 데이터)
SELECT customer_id, name, phone
FROM analytics.bronze.customers
WHERE phone IS NULL
LIMIT 5;

-- SDP 파이프라인에서 Silver 변환 정의

-- 고객 테이블: CDC 이력에서 최신 상태만 유지 (SCD Type 1)
CREATE OR REFRESH STREAMING TABLE analytics.silver.customers;

APPLY CHANGES INTO analytics.silver.customers
FROM STREAM(analytics.bronze.customers)
KEYS (customer_id)
SEQUENCE BY _commit_timestamp
STORED AS SCD TYPE 1;

-- 주문 테이블: 데이터 품질 검증 + 정제
CREATE OR REFRESH STREAMING TABLE analytics.silver.orders (
    CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
    CONSTRAINT valid_customer EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS
SELECT
    order_id,
    customer_id,
    amount,
    order_date,
    _ingested_at AS ingested_at
FROM STREAM(analytics.bronze.orders)
WHERE _change_type != 'DELETE';

# 파이프라인 상태 확인 (Databricks CLI)
# databricks pipelines get --pipeline-id <pipeline-id>

# 파이프라인이 FAILED 상태인 경우:
# 1. Pipeline UI에서 에러 메시지 확인
# 2. 원인 해결 (위의 표 참고)
# 3. Pipeline UI에서 "Start" 클릭하여 재시작
#    - "Full Refresh": 초기 스냅샷부터 다시 시작
#    - "Start": 마지막 체크포인트에서 이어서 시작

-- 1. 파이프라인 중지 (UI에서 Stop 클릭 또는 API 사용)

-- 2. 수집된 테이블 삭제
DROP TABLE IF EXISTS analytics.bronze.customers;
DROP TABLE IF EXISTS analytics.bronze.orders;
DROP TABLE IF EXISTS analytics.silver.customers;
DROP TABLE IF EXISTS analytics.silver.orders;

-- 3. Connection 삭제
DROP CONNECTION IF EXISTS mysql_ecommerce;

-- 4. 스키마/카탈로그 삭제 (필요 시)
-- DROP SCHEMA IF EXISTS analytics.bronze CASCADE;
-- DROP SCHEMA IF EXISTS analytics.silver CASCADE;
-- DROP CATALOG IF EXISTS analytics CASCADE;