Databricks

# Databricks 노트북 예제: Delta Lake + Unity Catalog 활용

# 1. Unity Catalog에서 카탈로그/스키마 설정
spark.sql("USE CATALOG main")
spark.sql("USE SCHEMA sales_analytics")

# 2. Delta Lake 테이블 생성 (Unity Catalog 관리)
spark.sql("""
CREATE TABLE IF NOT EXISTS bronze_orders (
    order_id STRING,
    customer_id STRING,
    product_id STRING,
    quantity INT,
    price DECIMAL(10, 2),
    order_date TIMESTAMP,
    _ingest_timestamp TIMESTAMP
)
USING DELTA
PARTITIONED BY (DATE(order_date))
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true'
)
""")

# 3. Auto Loader로 S3/ADLS에서 데이터 스트리밍 수집
raw_orders = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/orders_schema")
    .load("s3://data-lake/raw/orders/"))

# 변환 및 Bronze 테이블에 저장
from pyspark.sql.functions import current_timestamp

bronze_stream = (raw_orders
    .withColumn("_ingest_timestamp", current_timestamp())
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/bronze_orders")
    .trigger(availableNow=True)
    .toTable("bronze_orders"))

# 4. Delta Lake Time Travel - 과거 버전 조회
# 특정 버전으로 조회
df_v5 = spark.read.format("delta").option("versionAsOf", 5).table("bronze_orders")

# 특정 시점으로 조회
df_yesterday = spark.read.format("delta").option("timestampAsOf", "2024-01-15").table("bronze_orders")

# 5. Unity Catalog - 테이블 권한 관리
spark.sql("GRANT SELECT ON TABLE bronze_orders TO `data-analysts@company.com`")
spark.sql("GRANT MODIFY ON TABLE bronze_orders TO `data-engineers@company.com`")

# 6. 데이터 리니지 확인 (Unity Catalog)
# Databricks UI에서 테이블 -> Lineage 탭에서 시각적으로 확인 가능

print("Delta 테이블 히스토리:")
display(spark.sql("DESCRIBE HISTORY bronze_orders"))