CI/CD 패턴과 거버넌스

# 1단계: 모델 학습 및 challenger 등록
with mlflow.start_run():
    model = train_model(training_data)
    mlflow.sklearn.log_model(model, "model",
        registered_model_name="catalog.schema.fraud_detection")

# 새 버전에 challenger alias 부여
latest_version = client.get_latest_versions("catalog.schema.fraud_detection")[0].version
client.set_registered_model_alias(
    "catalog.schema.fraud_detection", "challenger", latest_version
)

# 2단계: 자동 검증 (Validation Job)
challenger = mlflow.pyfunc.load_model("models:/catalog.schema.fraud_detection@challenger")
champion = mlflow.pyfunc.load_model("models:/catalog.schema.fraud_detection@champion")

# 검증 데이터셋으로 성능 비교
challenger_metrics = evaluate_model(challenger, validation_data)
champion_metrics = evaluate_model(champion, validation_data)

# 3단계: 프로모션 판단
if challenger_metrics["f1_score"] > champion_metrics["f1_score"] * 1.02:  # 2% 이상 개선
    # 현재 champion을 rollback으로 보존
    champion_version = client.get_model_version_by_alias(
        "catalog.schema.fraud_detection", "champion"
    ).version
    client.set_registered_model_alias(
        "catalog.schema.fraud_detection", "rollback", champion_version
    )

    # challenger를 champion으로 승격
    client.set_registered_model_alias(
        "catalog.schema.fraud_detection", "champion", latest_version
    )
    print(f"✅ Promoted v{latest_version} to champion (F1: {challenger_metrics['f1_score']:.4f})")
else:
    print(f"❌ Challenger did not pass threshold (F1: {challenger_metrics['f1_score']:.4f} vs {champion_metrics['f1_score']:.4f})")

# Model Serving에서 트래픽 분할 A/B 테스트
w.serving_endpoints.update_config(
    name="fraud-detection",
    served_entities=[
        {
            "entity_name": "catalog.schema.fraud_detection",
            "entity_version": "champion",     # 기존 모델: 90% 트래픽
            "workload_size": "Small",
            "scale_to_zero_enabled": True,
            "traffic_percentage": 90
        },
        {
            "entity_name": "catalog.schema.fraud_detection",
            "entity_version": "challenger",   # 새 모델: 10% 트래픽
            "workload_size": "Small",
            "scale_to_zero_enabled": True,
            "traffic_percentage": 10
        }
    ]
)

-- 카탈로그 레벨: ML 카탈로그 전체에 대한 기본 권한
GRANT USE CATALOG ON CATALOG ml_catalog TO `all_data_scientists`;

-- 스키마 레벨: 모델 생성 권한
GRANT CREATE MODEL ON SCHEMA ml_catalog.production TO `senior_ml_engineers`;

-- 모델 레벨: 개별 모델 사용 권한
GRANT EXECUTE ON MODEL ml_catalog.production.fraud_detection TO `serving_service_principal`;

-- 읽기 전용 (메타데이터 조회만)
GRANT SELECT ON MODEL ml_catalog.production.fraud_detection TO `ml_auditors`;

-- 모델 접근 감사 로그 확인
SELECT
    event_time,
    user_identity.email AS user,
    action_name,
    request_params.full_name_arg AS model_name,
    request_params.version AS model_version,
    response.status_code
FROM system.access.audit
WHERE service_name = 'unityCatalog'
  AND action_name IN ('getRegisteredModel', 'createModelVersion', 'setModelAlias')
  AND request_params.full_name_arg LIKE '%fraud_detection%'
ORDER BY event_time DESC
LIMIT 50;

# 모델 승인 상태 관리
def request_approval(model_name, version):
    """모델 배포 승인 요청"""
    client.set_model_version_tag(model_name, version, "approval_status", "pending")
    client.set_model_version_tag(model_name, version, "requested_by", "kim@company.com")
    client.set_model_version_tag(model_name, version, "requested_at", str(datetime.now()))
    # Slack 알림 전송 등

def approve_model(model_name, version, approver):
    """모델 배포 승인"""
    client.set_model_version_tag(model_name, version, "approval_status", "approved")
    client.set_model_version_tag(model_name, version, "approved_by", approver)
    client.set_model_version_tag(model_name, version, "approved_at", str(datetime.now()))

def reject_model(model_name, version, reviewer, reason):
    """모델 배포 거부"""
    client.set_model_version_tag(model_name, version, "approval_status", "rejected")
    client.set_model_version_tag(model_name, version, "rejected_by", reviewer)
    client.set_model_version_tag(model_name, version, "rejection_reason", reason)

# Dev → Staging 프로모션
def promote_to_staging(dev_model_name, dev_version, staging_model_name):
    """개발 환경에서 스테이징으로 모델 복사"""
    source_uri = f"models:/{dev_model_name}/{dev_version}"

    mlflow.set_registry_uri("databricks-uc")
    result = mlflow.register_model(source_uri, staging_model_name)

    client.set_registered_model_alias(
        staging_model_name, "challenger", result.version
    )

    # 태그에 출처 기록
    client.set_model_version_tag(
        staging_model_name, result.version,
        "source", f"{dev_model_name}/v{dev_version}"
    )
    return result.version

# Staging → Prod 프로모션 (승인 후)
def promote_to_prod(staging_model_name, staging_version, prod_model_name):
    """스테이징에서 프로덕션으로 모델 복사 (승인 필요)"""
    # 승인 상태 확인
    version_info = client.get_model_version(staging_model_name, staging_version)
    tags = {t.key: t.value for t in version_info.tags}

    if tags.get("approval_status") != "approved":
        raise ValueError("모델이 승인되지 않았습니다")

    source_uri = f"models:/{staging_model_name}/{staging_version}"
    result = mlflow.register_model(source_uri, prod_model_name)

    # 현재 champion을 rollback으로 보존
    try:
        current = client.get_model_version_by_alias(prod_model_name, "champion")
        client.set_registered_model_alias(prod_model_name, "rollback", current.version)
    except:
        pass

    # 새 버전을 champion으로 설정
    client.set_registered_model_alias(prod_model_name, "champion", result.version)
    return result.version

-- 특정 모델이 어떤 테이블에서 학습되었는지
SELECT
    source_table_full_name,
    source_type
FROM system.lineage.table_lineage
WHERE target_table_full_name = 'catalog.ml.fraud_detection'
  AND target_type = 'MODEL';

-- 특정 테이블이 변경되면 영향받는 모델
SELECT
    target_table_full_name AS affected_model,
    target_type
FROM system.lineage.table_lineage
WHERE source_table_full_name = 'catalog.ecommerce.gold_orders'
  AND target_type = 'MODEL';