Genie Code 활용 시나리오

@customers 테이블에 대해 고객 세그멘테이션 분석을 수행해줘.
K-means 클러스터링을 사용하고 MLflow에 실험을 기록해줘.

사용자: @customers 테이블에서 고객 이탈 예측 모델을 만들어줘.
       target 컬럼은 'churned'야.

Genie: 고객 이탈 예측 모델을 구축하겠습니다.
       [계획]
       1. 데이터 로드 및 EDA
       2. 피처 엔지니어링
       3. 학습/테스트 데이터 분리
       4. 여러 모델 학습 (LogisticRegression, RandomForest, XGBoost)
       5. 모델 비교 및 최적 모델 선택
       6. MLflow에 실험 로깅

       실행을 시작합니다...

       [셀 1] 데이터 로드 — 50,000행, 23개 컬럼 확인
       [셀 2] EDA — churned 비율: 18.3%, 클래스 불균형 발견
       [셀 3] 피처 엔지니어링 — 15개 피처 선택, SMOTE로 오버샘플링
       [셀 4] 모델 학습 결과:
         - LogisticRegression: AUC 0.82
         - RandomForest: AUC 0.89
         - XGBoost: AUC 0.91 ← 최적
       [셀 5] XGBoost 모델을 MLflow에 등록했습니다.

사용자: 피처 중요도를 시각화해줘.

Genie: XGBoost 모델의 피처 중요도 차트를 생성합니다...
       [차트] 상위 10개 피처: contract_length, monthly_charges, tenure...

raw_orders와 raw_customers 테이블을 조인하여
daily_order_summary 테이블을 업데이트하는 SDP 파이프라인을 만들어줘.
데이터 품질 검증도 포함해줘.

사용자: raw_orders에서 daily_order_summary까지의 SDP 파이프라인을 만들어줘.

Genie: Lakeflow SDP 파이프라인을 생성하겠습니다.

       [생성된 코드]

       # Bronze 레이어: Auto Loader로 원본 데이터 수집
       @dlt.table(comment="원본 주문 데이터")
       def raw_orders():
           return spark.readStream.format("cloudFiles")...

       # Silver 레이어: 데이터 품질 검증
       @dlt.table(comment="정제된 주문 데이터")
       @dlt.expect_or_drop("valid_amount", "amount > 0")
       @dlt.expect_or_drop("valid_date", "order_date IS NOT NULL")
       def cleaned_orders():
           return dlt.read("raw_orders").join(...)

       # Gold 레이어: 일별 요약
       @dlt.table(comment="일별 주문 요약")
       def daily_order_summary():
           return dlt.read("cleaned_orders")
               .groupBy("order_date", "region")
               .agg(...)

사용자: 데이터 품질 규칙을 더 추가해줘. 주문 금액은 0보다 크고 100만 이하여야 해.

Genie: 데이터 품질 규칙을 업데이트합니다...
       @dlt.expect_or_drop("valid_amount_range", "amount > 0 AND amount <= 1000000")

@sales_data를 분석하고 월별 매출 추이, 지역별 성과,
상위 제품을 보여주는 대시보드를 만들어줘.

사용자: 이 대시보드 쿼리가 느려. 최적화해줘.

Genie: 쿼리를 분석한 결과, 다음 최적화를 적용합니다:

       1. 서브쿼리 → CTE(Common Table Expression)로 변환
       2. 불필요한 DISTINCT 제거
       3. WHERE 절에 파티션 프루닝 조건 추가
       4. 집계 전 필터링으로 처리 데이터량 감소

       예상 성능 개선: 쿼리 시간 70% 감소

이 RAG 체인의 evaluation 결과를 분석하고
응답 품질이 낮은 케이스를 식별해줘.

사용자: 이 RAG 체인에서 faithfulness 점수가 낮은 케이스를 분석해줘.

Genie: MLflow experiment에서 평가 결과를 로드합니다...

       전체 200개 테스트 중 faithfulness < 0.5인 케이스: 23개 (11.5%)

       원인 분류:
       - 검색된 문서가 질문과 무관: 12건 (52%)
       - 검색 결과는 맞지만 LLM이 환각: 7건 (30%)
       - 컨텍스트 길이 초과로 잘림: 4건 (18%)

       주요 개선 제안:
       1. 임베딩 모델 변경 (현재 e5-small → e5-large)
       2. 청크 크기 조정 (현재 500 → 300 토큰)
       3. Retrieval에 reranker 추가

사용자: 검색 실패 케이스의 trace를 자세히 보여줘.

Genie: 검색 실패 12건의 MLflow Trace를 분석합니다...

       [Trace 1] 질문: "반품 정책은?"
         → Retriever: "배송 정책" 문서 3개 반환 (관련성 0.3)
         → 원인: "반품"과 "배송"이 같은 임베딩 공간에 위치
         → 제안: 반품 관련 문서에 키워드 메타데이터 추가

# 패턴 1: 전체 평가 결과 분석
"MLflow experiment '{experiment_name}'의 evaluation 결과를 분석해줘.
 각 평가 지표(faithfulness, relevance, toxicity)의 분포를 시각화하고,
 개선이 필요한 영역을 우선순위별로 정리해줘."

# 패턴 2: 특정 실패 케이스 심층 분석
"faithfulness 점수가 0.3 이하인 케이스를 모아서,
 각각의 MLflow Trace를 분석하고,
 실패 원인을 카테고리별로 분류해줘."

# 패턴 3: 개선 전후 비교
"실험 A와 실험 B의 evaluation 결과를 비교해줘.
 어떤 지표가 개선되었고, 어떤 케이스가 새로 실패했는지 분석해줘."

사용자: @catalog.schema.orders 테이블의 데이터 품질을 종합적으로 점검해줘.
       null 비율, 중복 레코드, 날짜 유효성, 금액 범위를 체크하고,
       이상이 있으면 요약 보고서를 만들어줘.

Genie: 데이터 품질 점검을 시작합니다.

       [셀 1] 기본 통계 — 1,234,567행, 18컬럼, 최종 업데이트: 2시간 전
       [셀 2] Null 분석 — customer_email 12.3% null (경고), 나머지 정상
       [셀 3] 중복 분석 — order_id 기준 중복 0건 (정상)
       [셀 4] 날짜 유효성 — 미래 날짜 23건 발견 (경고)
       [셀 5] 금액 범위 — 음수 금액 5건, 1억 초과 2건 (경고)
       [셀 6] 품질 보고서:
         ✅ 정상: 중복 없음, 대부분 컬럼 null 비율 0%
         ⚠️ 경고: customer_email null 12.3%
         ❌ 이상: 미래 날짜 23건, 비정상 금액 7건

사용자: 다음 pandas 코드를 PySpark로 변환해줘.
       분산 처리에 최적화된 방식으로 작성하고,
       pandas에서는 불필요했지만 PySpark에서 중요한
       파티셔닝이나 캐싱도 필요하면 추가해줘.

Genie: Pandas → PySpark 변환을 수행합니다.

       변환 시 적용한 최적화:
       1. df.apply() → PySpark 내장 함수(when, coalesce)로 대체
       2. merge() → broadcast join으로 변환 (작은 테이블 기준)
       3. groupby().agg() → PySpark의 groupBy().agg()로 변환
       4. 날짜 파티셔닝 추가 (dt 컬럼 기준)
       5. 중간 결과에 .cache() 추가 (3회 이상 참조)