Delta Lake 핵심

-- Delta 테이블 생성
CREATE TABLE catalog.schema.customers (
    customer_id BIGINT,
    name STRING,
    email STRING,
    city STRING,
    signup_date DATE
) USING DELTA;

-- 데이터 삽입
INSERT INTO catalog.schema.customers VALUES
(1, '김철수', 'cs.kim@email.com', '서울', '2025-01-15'),
(2, '이영희', 'yh.lee@email.com', '부산', '2025-02-20'),
(3, '박민수', 'ms.park@email.com', '대구', '2025-03-10');

-- 현재 데이터 조회
SELECT * FROM catalog.schema.customers;

-- 데이터 수정
UPDATE catalog.schema.customers
SET city = '인천'
WHERE customer_id = 1;

-- 수정 전 데이터 조회 (버전 번호로)
SELECT * FROM catalog.schema.customers VERSION AS OF 0;

-- 수정 전 데이터 조회 (타임스탬프로)
SELECT * FROM catalog.schema.customers TIMESTAMP AS OF '2025-03-15 10:00:00';

-- 테이블 변경 이력 확인
DESCRIBE HISTORY catalog.schema.customers;

-- customers 테이블에 잘못된 데이터 삽입 시도
INSERT INTO catalog.schema.customers VALUES
(4, '최지은', 'je.choi@email.com', '서울', 'not-a-date');
-- ❌ 오류 발생: 'not-a-date'는 DATE 타입이 아닙니다

-- 새 컬럼 추가
ALTER TABLE catalog.schema.customers
ADD COLUMN phone_number STRING;

-- 기존 데이터는 새 컬럼이 NULL로 채워짐
SELECT * FROM catalog.schema.customers;
-- +------------+------+------------------+----+------------+--------------+
-- |customer_id | name | email            |city|signup_date | phone_number |
-- +------------+------+------------------+----+------------+--------------+
-- |1           |김철수|cs.kim@email.com  |인천|2025-01-15  | NULL         |
-- |2           |이영희|yh.lee@email.com  |부산|2025-02-20  | NULL         |
-- ...

-- 1. Delta 테이블 생성
CREATE TABLE catalog.schema.products (
    product_id BIGINT,
    name STRING,
    category STRING,
    price DECIMAL(10, 2),
    stock INT
) USING DELTA;

-- 2. 데이터 삽입
INSERT INTO catalog.schema.products VALUES
(101, '무선 키보드', '전자제품', 89000.00, 150),
(102, '블루투스 이어폰', '전자제품', 65000.00, 300),
(103, '텀블러 500ml', '생활용품', 25000.00, 500);

-- 3. 데이터 수정 (UPDATE)
UPDATE catalog.schema.products
SET price = 79000.00, stock = stock - 10
WHERE product_id = 101;

-- 4. 변경 이력 확인
DESCRIBE HISTORY catalog.schema.products;

-- 5. 수정 전 데이터 확인 (타임 트래블)
SELECT * FROM catalog.schema.products VERSION AS OF 1;

-- 6. 테이블 상세 정보 확인
DESCRIBE DETAIL catalog.schema.products;

// _delta_log/00005.json 예시 (단순화)
{
  "add": {
    "path": "part-00003-abc123.parquet",
    "size": 52428800,
    "modificationTime": 1711900800000,
    "dataChange": true,
    "stats": "{\"numRecords\":125000,\"minValues\":{\"order_date\":\"2025-01-01\"},\"maxValues\":{\"order_date\":\"2025-03-31\"},\"nullCount\":{\"order_date\":0}}"
  }
}
{
  "remove": {
    "path": "part-00001-def456.parquet",
    "deletionTimestamp": 1711900800000,
    "dataChange": true
  }
}
{
  "commitInfo": {
    "operation": "MERGE",
    "operationParameters": {"predicate": "t.id = s.id"},
    "readVersion": 4,
    "operationMetrics": {"numTargetRowsUpdated": "3200", "numOutputRows": "128000"}
  }
}

-- 이 쿼리는 order_date 범위에 해당하는 파일만 읽습니다
SELECT * FROM catalog.schema.orders
WHERE order_date BETWEEN '2025-03-01' AND '2025-03-31';
-- Stats에서 max(order_date) < '2025-03-01'인 파일은 자동으로 건너뜀

Writer A: 버전 5 읽기 → 변환 처리 (수 분) → 버전 6으로 커밋 시도 → ✅ 성공
Writer B: 버전 5 읽기 → 변환 처리 (수 분) → 버전 6으로 커밋 시도 → ❌ 충돌!
         → 버전 6 로그를 읽고 충돌 여부 판단 → 재시도 or ConflictException

from delta.exceptions import ConcurrentAppendException, ConcurrentDeleteReadException
import time

MAX_RETRIES = 3
for attempt in range(MAX_RETRIES):
    try:
        # MERGE 작업 실행
        deltaTable.alias("t").merge(
            source.alias("s"),
            "t.id = s.id"
        ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
        break  # 성공 시 루프 종료
    except ConcurrentAppendException:
        if attempt < MAX_RETRIES - 1:
            time.sleep(2 **attempt)  # 지수 백오프
            continue
        raise  # 최대 재시도 초과 시 예외 전파

-- 테이블의 격리 수준을 Serializable로 변경
ALTER TABLE catalog.schema.financial_ledger
SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable');

_delta_log/
  00000000000000000000.json     ← 버전 0
  00000000000000000001.json     ← 버전 1
  ...
  00000000000000000010.checkpoint.parquet  ← 체크포인트 (v0~v10 요약)
  00000000000000000011.json     ← 버전 11 (체크포인트 이후 증분)
  ...
  00000000000000000020.checkpoint.parquet  ← 체크포인트 (v0~v20 요약)

-- 로그 보관 기간 조정 (기본 30일 → 7일)
ALTER TABLE catalog.schema.huge_table
SET TBLPROPERTIES (
  'delta.logRetentionDuration' = 'interval 7 days',
  'delta.checkpointInterval' = '10'  -- 기본값 유지 권장
);

-- 오래된 파일 정리 (7일 이전)
VACUUM catalog.schema.huge_table RETAIN 168 HOURS;