Databricks AI Engineer (FDE) 합격 가이드: Spark, Unity Catalog, RAG부터 고객 배포까지

# RDD (2014) - 저수준 API, 타입 안전하지만 최적화 어려움
rdd = sc.parallelize([1, 2, 3, 4, 5])
result = rdd.filter(lambda x: x > 2).map(lambda x: x * 2).collect()

# DataFrame (2015) - SQL 친화적, Catalyst Optimizer 활용
df = spark.read.parquet("s3://data/events/")
result = df.filter(df.age > 25).groupBy("city").count()

# Dataset (Scala only) - DataFrame + 컴파일 타임 타입 안전성
# Python에서는 DataFrame이 곧 Dataset[Row]

# Catalyst의 최적화 단계를 확인하는 방법
df = spark.read.parquet("s3://data/sales/")
optimized = df.filter("amount > 1000").join(
    spark.read.parquet("s3://data/customers/"),
    "customer_id"
)

# 논리적/물리적 실행 계획 확인
optimized.explain(True)

# 출력 예시:
# == Parsed Logical Plan ==
# == Analyzed Logical Plan ==
# == Optimized Logical Plan ==   <-- Predicate Pushdown, Column Pruning 적용됨
# == Physical Plan ==             <-- Join 전략(BroadcastHashJoin 등) 결정됨

# 파티셔닝 전략 - 데이터 분포에 따른 최적화
# 1. Hash Partitioning (기본)
df.repartition(200, "customer_id")

# 2. Range Partitioning - 정렬된 데이터에 유리
df.repartitionByRange(200, "event_date")

# 3. 셔플 최적화 - broadcast join 사용
from pyspark.sql.functions import broadcast

# 작은 테이블(10MB 이하)을 브로드캐스트하여 셔플 제거
result = large_df.join(broadcast(small_df), "key")

# 4. Salting - 데이터 스큐 해결
from pyspark.sql.functions import concat, lit, rand, floor

# 핫키에 소금(salt) 추가하여 균등 분배
salt_range = 10
skewed_df = skewed_df.withColumn(
    "salted_key",
    concat("key", lit("_"), floor(rand() * salt_range).cast("string"))
)

# AQE (Adaptive Query Execution) - Spark 3.0+
# spark.conf 설정
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# PySpark - 데이터 과학자와의 협업에 유리
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, sum as spark_sum

spark = SparkSession.builder.appName("etl").getOrCreate()

df = (
    spark.read.format("delta")
    .load("s3://lakehouse/bronze/events")
    .filter(col("event_date") >= "2025-01-01")
    .withColumn("category",
        when(col("amount") > 1000, "high")
        .when(col("amount") > 100, "medium")
        .otherwise("low")
    )
    .groupBy("category")
    .agg(spark_sum("amount").alias("total_amount"))
)

// Scala Spark - 타입 안전성, 더 나은 성능
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder.appName("etl").getOrCreate()
import spark.implicits._

val df = spark.read.format("delta")
  .load("s3://lakehouse/bronze/events")
  .filter($"event_date" >= "2025-01-01")
  .withColumn("category",
    when($"amount" > 1000, "high")
    .when($"amount" > 100, "medium")
    .otherwise("low")
  )
  .groupBy("category")
  .agg(sum("amount").alias("total_amount"))

# Delta Lake 기본 CRUD
# 1. 테이블 생성
df.write.format("delta").mode("overwrite").saveAsTable("lakehouse.bronze.raw_events")

# 2. 업데이트 (MERGE/UPSERT)
from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "lakehouse.silver.customers")
source = spark.read.format("delta").load("s3://staging/new_customers/")

target.alias("t").merge(
    source.alias("s"),
    "t.customer_id = s.customer_id"
).whenMatchedUpdate(set={
    "name": "s.name",
    "email": "s.email",
    "updated_at": "current_timestamp()"
}).whenNotMatchedInsertAll().execute()

# 3. Time Travel - 과거 버전 조회
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load(path)
df_yesterday = spark.read.format("delta").option("timestampAsOf", "2025-03-22").load(path)

# 4. RESTORE - 테이블을 과거 버전으로 복원
spark.sql("RESTORE TABLE lakehouse.silver.customers VERSION AS OF 5")

# Bronze Layer - 원본 데이터 그대로 적재 (Append-only)
raw_df = (
    spark.readStream
    .format("cloudFiles")  # Auto Loader
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://schema/bronze/events/")
    .load("s3://raw-data/events/")
)

bronze_df = raw_df.withColumn("_ingested_at", current_timestamp())

bronze_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://checkpoints/bronze/events/") \
    .trigger(availableNow=True) \
    .toTable("lakehouse.bronze.raw_events")

# Silver Layer - 정제된 데이터 (Dedup, Cleansing, Type Casting)
silver_df = (
    spark.readStream.table("lakehouse.bronze.raw_events")
    .dropDuplicates(["event_id"])
    .filter(col("event_type").isNotNull())
    .withColumn("amount", col("amount").cast("decimal(18,2)"))
    .withColumn("event_date", to_date(col("event_timestamp")))
)

silver_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://checkpoints/silver/events/") \
    .trigger(availableNow=True) \
    .toTable("lakehouse.silver.clean_events")

# Gold Layer - 비즈니스 집계 테이블
gold_df = (
    spark.read.table("lakehouse.silver.clean_events")
    .groupBy("customer_id", "event_date")
    .agg(
        count("*").alias("event_count"),
        spark_sum("amount").alias("daily_total"),
        avg("amount").alias("avg_amount")
    )
)

gold_df.write.format("delta").mode("overwrite") \
    .saveAsTable("lakehouse.gold.customer_daily_summary")

-- CDF 활성화
ALTER TABLE lakehouse.silver.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- CDF 조회: 변경된 레코드만 추출
SELECT * FROM table_changes('lakehouse.silver.customers', 5)
WHERE _change_type IN ('insert', 'update_postimage');

-- Liquid Clustering 설정
CREATE TABLE lakehouse.silver.events
CLUSTER BY (event_date, customer_id)
AS SELECT * FROM lakehouse.bronze.raw_events;

-- OPTIMIZE가 자동으로 최적 클러스터링 적용
OPTIMIZE lakehouse.silver.events;

-- 카탈로그 생성
CREATE CATALOG production;
CREATE CATALOG development;

-- 스키마(데이터베이스) 생성
CREATE SCHEMA production.finance;
CREATE SCHEMA production.marketing;

-- 테이블 참조
SELECT * FROM production.finance.transactions
WHERE transaction_date >= '2025-01-01';

-- 크로스 카탈로그 조인
SELECT a.*, b.segment
FROM production.finance.transactions a
JOIN production.marketing.customer_segments b
ON a.customer_id = b.customer_id;

# 리니지는 자동 추적되므로 별도 코드 불필요
# UI에서 확인하거나 REST API로 조회 가능

# Unity Catalog REST API로 리니지 조회
import requests

response = requests.get(
    "https://workspace.cloud.databricks.com/api/2.1/unity-catalog/lineage/table-lineage",
    headers=headers,
    params={"table_name": "production.finance.transactions"}
)

-- Column Level Security: 마스킹 함수 정의
CREATE FUNCTION production.finance.mask_ssn(ssn STRING)
RETURNS STRING
RETURN CASE
    WHEN IS_ACCOUNT_GROUP_MEMBER('finance_admins') THEN ssn
    ELSE CONCAT('***-**-', RIGHT(ssn, 4))
END;

-- 컬럼에 마스킹 함수 적용
ALTER TABLE production.finance.customers
ALTER COLUMN ssn SET MASK production.finance.mask_ssn;

-- Row Level Security: 행 필터 함수 정의
CREATE FUNCTION production.finance.region_filter(region STRING)
RETURNS BOOLEAN
RETURN CASE
    WHEN IS_ACCOUNT_GROUP_MEMBER('global_admins') THEN TRUE
    WHEN IS_ACCOUNT_GROUP_MEMBER('apac_team') AND region = 'APAC' THEN TRUE
    ELSE FALSE
END;

ALTER TABLE production.finance.transactions
SET ROW FILTER production.finance.region_filter ON (region);

-- 태그 기반 접근 제어
ALTER TABLE production.finance.transactions SET TAGS ('pii' = 'true', 'classification' = 'confidential');

-- 태그 기반 정책
GRANT SELECT ON TABLE production.finance.transactions
TO `data_analysts`
WHERE TAGS('classification') != 'confidential';

import mlflow
from mlflow.tracking import MlflowClient

# 실험 생성 및 실행
mlflow.set_experiment("/Shared/customer-churn-prediction")

with mlflow.start_run(run_name="xgboost_v2") as run:
    # 하이퍼파라미터 로깅
    mlflow.log_param("max_depth", 6)
    mlflow.log_param("learning_rate", 0.1)
    mlflow.log_param("n_estimators", 200)

    # 모델 학습
    model = xgb.XGBClassifier(max_depth=6, learning_rate=0.1, n_estimators=200)
    model.fit(X_train, y_train)

    # 메트릭 로깅
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("f1_score", f1_score(y_test, predictions))

    # 모델 아티팩트 로깅
    mlflow.xgboost.log_model(model, "model")

    # Unity Catalog에 모델 등록
    mlflow.register_model(
        f"runs:/{run.info.run_id}/model",
        "production.ml_models.churn_predictor"
    )

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 피처 테이블 생성
fe.create_table(
    name="production.features.customer_features",
    primary_keys=["customer_id"],
    timestamp_keys=["event_date"],
    df=customer_features_df,
    description="고객 행동 피처: 최근 구매 빈도, 평균 구매 금액, 마지막 로그인 일수"
)

# 피처 룩업으로 학습 데이터 생성
from databricks.feature_engineering import FeatureLookup

training_set = fe.create_training_set(
    df=label_df,
    feature_lookups=[
        FeatureLookup(
            table_name="production.features.customer_features",
            lookup_key="customer_id",
            timestamp_lookup_key="event_date"
        )
    ],
    label="churned"
)

training_df = training_set.load_df()

# Autolog 활성화 - 하이퍼파라미터, 메트릭, 모델을 자동 추적
mlflow.autolog()

# 이후 모든 학습이 자동으로 MLflow에 기록됨
from sklearn.ensemble import RandomForestClassifier
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X_train, y_train)
# 자동으로: params, metrics, model artifact, feature importance 등이 기록됨

import requests
import json

# Model Serving Endpoint 생성 (REST API)
endpoint_config = {
    "name": "churn-predictor-endpoint",
    "config": {
        "served_entities": [{
            "entity_name": "production.ml_models.churn_predictor",
            "entity_version": "3",
            "workload_size": "Small",
            "scale_to_zero_enabled": True
        }],
        "auto_capture_config": {
            "catalog_name": "production",
            "schema_name": "ml_monitoring",
            "table_name_prefix": "churn_predictor"
        }
    }
}

# 엔드포인트에 추론 요청
response = requests.post(
    "https://workspace.cloud.databricks.com/serving-endpoints/churn-predictor-endpoint/invocations",
    headers=headers,
    json={"dataframe_records": [{"customer_id": "C001", "purchase_count": 12, "avg_amount": 150.0}]}
)

from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()

# Vector Search 엔드포인트 생성
vsc.create_endpoint(name="rag-endpoint", endpoint_type="STANDARD")

# Delta Sync Index 생성 - Delta 테이블 변경 시 자동 동기화
vsc.create_delta_sync_index(
    endpoint_name="rag-endpoint",
    index_name="production.rag.document_index",
    source_table_name="production.rag.documents",
    pipeline_type="TRIGGERED",
    primary_key="doc_id",
    embedding_source_column="content",
    embedding_model_endpoint_name="databricks-bge-large-en"
)

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Step 1: Document Chunking
loader = PyPDFLoader("dbfs:/documents/product_manual.pdf")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = text_splitter.split_documents(documents)

# Step 2: Embedding (Databricks Foundation Model 사용)
# 청크를 Delta 테이블에 저장하면 Vector Search Index가 자동 임베딩
chunks_df = spark.createDataFrame([
    {"doc_id": f"doc_{i}", "content": chunk.page_content, "metadata": str(chunk.metadata)}
    for i, chunk in enumerate(chunks)
])
chunks_df.write.format("delta").mode("append").saveAsTable("production.rag.documents")

# Step 3: Search (유사도 검색)
results = vsc.get_index("rag-endpoint", "production.rag.document_index").similarity_search(
    query_text="제품 보증 정책은 무엇인가요?",
    columns=["content", "metadata"],
    num_results=5
)

# Step 4: Generate (LLM으로 답변 생성)
import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")

context = "\n".join([r["content"] for r in results["result"]["data_array"]])
prompt = f"""다음 컨텍스트를 기반으로 질문에 답하세요.

컨텍스트:
{context}

질문: 제품 보증 정책은 무엇인가요?
답변:"""

response = client.predict(
    endpoint="databricks-meta-llama-3-1-70b-instruct",
    inputs={"prompt": prompt, "max_tokens": 500, "temperature": 0.1}
)

from databricks.agents import Agent, AgentTool

# 에이전트 도구 정의
search_tool = AgentTool(
    name="document_search",
    description="사내 문서에서 관련 정보를 검색합니다",
    func=lambda query: vsc.get_index("rag-endpoint", "production.rag.document_index")
        .similarity_search(query_text=query, num_results=3)
)

sql_tool = AgentTool(
    name="data_query",
    description="데이터베이스에서 비즈니스 데이터를 조회합니다",
    func=lambda query: spark.sql(query).toPandas().to_dict()
)

# 에이전트 생성
agent = Agent(
    model="databricks-meta-llama-3-1-70b-instruct",
    tools=[search_tool, sql_tool],
    system_prompt="당신은 고객 지원 에이전트입니다. 문서 검색과 데이터 조회를 통해 정확한 답변을 제공하세요."
)

# Model Gateway - 여러 LLM 프로바이더를 단일 인터페이스로 통합
# Databricks UI에서 설정하거나 REST API 사용

# OpenAI 모델을 Databricks Gateway를 통해 호출
response = client.predict(
    endpoint="openai-gpt4-gateway",
    inputs={"messages": [{"role": "user", "content": "분석 보고서를 작성해주세요"}]}
)

# Anthropic Claude를 Gateway를 통해 호출
response = client.predict(
    endpoint="anthropic-claude-gateway",
    inputs={"messages": [{"role": "user", "content": "코드를 리뷰해주세요"}]}
)

# 오픈소스 모델 (Llama, Mistral 등)도 동일 인터페이스
response = client.predict(
    endpoint="databricks-meta-llama-3-1-70b-instruct",
    inputs={"prompt": "데이터를 요약해주세요", "max_tokens": 300}
)

# 1. Available Now - 현재 가용한 데이터만 처리 후 종료 (배치-스트리밍 통합)
query = (
    df.writeStream
    .format("delta")
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# 2. Processing Time - 지정 간격마다 마이크로배치 실행
query = (
    df.writeStream
    .format("delta")
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# 3. Continuous (실험적) - 밀리초 지연시간의 실시간 처리
query = (
    df.writeStream
    .format("delta")
    .trigger(continuous="1 second")
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# 이벤트 시간 기반 윈도우 집계 + 워터마크
from pyspark.sql.functions import window

windowed_counts = (
    events_df
    .withWatermark("event_time", "10 minutes")  # 10분 지연까지 허용
    .groupBy(
        window("event_time", "5 minutes", "1 minute"),  # 5분 윈도우, 1분 슬라이드
        "device_type"
    )
    .count()
)

windowed_counts.writeStream \
    .format("delta") \
    .outputMode("append") \
    .trigger(processingTime="1 minute") \
    .option("checkpointLocation", checkpoint_path) \
    .toTable("lakehouse.gold.device_counts_5min")

import dlt
from pyspark.sql.functions import col, current_timestamp

# Bronze: 원본 데이터 수집
@dlt.table(
    name="bronze_events",
    comment="원본 이벤트 데이터",
    table_properties={"quality": "bronze"}
)
def bronze_events():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("s3://raw-data/events/")
    )

# Silver: 데이터 품질 검증 포함
@dlt.table(
    name="silver_events",
    comment="정제된 이벤트 데이터"
)
@dlt.expect_or_drop("valid_event_id", "event_id IS NOT NULL")
@dlt.expect_or_fail("valid_amount", "amount >= 0")
@dlt.expect("valid_email", "email RLIKE '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+$'")
def silver_events():
    return (
        dlt.read_stream("bronze_events")
        .dropDuplicates(["event_id"])
        .withColumn("processed_at", current_timestamp())
    )

# Gold: 비즈니스 집계
@dlt.table(
    name="gold_daily_summary",
    comment="일별 비즈니스 요약"
)
def gold_daily_summary():
    return (
        dlt.read("silver_events")
        .groupBy("event_date", "category")
        .agg(
            count("*").alias("total_events"),
            spark_sum("amount").alias("total_amount")
        )
    )

# Auto Loader - 클라우드 스토리지의 신규 파일을 자동 감지하여 수집
df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://schema/auto_loader/events/")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("s3://raw-data/events/")
)

# Auto Loader의 장점:
# 1. 파일 목록 관리 불필요 - 신규 파일 자동 감지
# 2. 스키마 추론 및 진화 - 새 컬럼 자동 추가
# 3. 정확히 한 번(exactly-once) 처리 보장
# 4. 수백만 파일에서도 효율적 (파일 알림 모드)

# AWS에서의 Databricks 아키텍처
# 1. 스토리지: S3 + Delta Lake
storage_config = {
    "data_bucket": "s3://company-lakehouse-prod/",
    "checkpoint_bucket": "s3://company-checkpoints-prod/",
    "metastore_bucket": "s3://company-unity-catalog/",
    "encryption": "SSE-KMS",
    "kms_key": "arn:aws:kms:us-east-1:123456789012:key/xxx"
}

# 2. 컴퓨트: EC2 인스턴스 기반 클러스터
cluster_config = {
    "spark_version": "14.3.x-scala2.12",
    "node_type_id": "i3.xlarge",  # 스토리지 최적화
    "driver_node_type_id": "i3.2xlarge",
    "num_workers": 8,
    "autoscale": {"min_workers": 4, "max_workers": 16},
    "aws_attributes": {
        "instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/databricks-role",
        "availability": "SPOT_WITH_FALLBACK",
        "spot_bid_price_percent": 100
    }
}

# Azure에서의 Databricks 아키텍처
# 1. 스토리지: ADLS Gen2 + Delta Lake
storage_config = {
    "container": "abfss://lakehouse@companystorage.dfs.core.windows.net/",
    "metastore": "abfss://unity-catalog@companystorage.dfs.core.windows.net/",
    "encryption": "Microsoft Managed Keys",
    "network": "Private Endpoint"
}

# 2. 네트워킹: VNet Injection
# Databricks 워크스페이스를 고객 VNet에 직접 배포
# - Private Link로 컨트롤 플레인 접근
# - No Public IP로 클러스터 운영
# - NSG(Network Security Group)로 트래픽 제어

# GCP에서의 Databricks 아키텍처
# 1. 스토리지: GCS + Delta Lake
storage_config = {
    "bucket": "gs://company-lakehouse-prod/",
    "metastore": "gs://company-unity-catalog/",
    "encryption": "Customer-Managed Encryption Key (CMEK)"
}

# 2. BigQuery와의 연동
# Databricks에서 BigQuery 테이블 직접 조회
bq_df = (
    spark.read
    .format("bigquery")
    .option("table", "project.dataset.table")
    .option("viewsEnabled", "true")
    .load()
)

# Terraform으로 Databricks 워크스페이스 + Unity Catalog 설정

provider "databricks" {
  alias = "workspace"
  host  = databricks_workspace.main.workspace_url
}

# 워크스페이스 생성 (AWS 예시)
resource "databricks_workspace" "main" {
  workspace_name = "production-lakehouse"
  region         = "us-east-1"

  aws_config {
    subnet_ids         = var.private_subnet_ids
    security_group_ids = [var.security_group_id]
    instance_profile   = var.instance_profile_arn
  }

  storage_config {
    s3_bucket_name = var.root_bucket_name
  }
}

# Unity Catalog Metastore
resource "databricks_metastore" "main" {
  provider      = databricks.workspace
  name          = "production-metastore"
  storage_root  = "s3://unity-catalog-metastore/"
  force_destroy = false
}

# 클러스터 정책
resource "databricks_cluster_policy" "data_engineering" {
  provider = databricks.workspace
  name     = "Data Engineering Policy"

  definition = jsonencode({
    "spark_version" : { "type" : "fixed", "value" : "14.3.x-scala2.12" },
    "autoscale.max_workers" : { "type" : "range", "maxValue" : 20 },
    "node_type_id" : { "type" : "allowlist", "values" : ["i3.xlarge", "i3.2xlarge"] },
    "custom_tags.team" : { "type" : "fixed", "value" : "data-engineering" }
  })
}