06. 배치 추론 (Batch Inference)

전체 노트북 코드: 06_batch_inference.py

​

이 노트북은 무엇을 하는가?

​

실시간 추론 vs 배치 추론

참고 배치 추론의 핵심은 “얼마나 빠르게 대량 예측을 하느냐”입니다. Spark UDF의 진짜 위력은 단일 모델을 클러스터 전체 노드에서 동시에 실행한다는 것입니다. 10만 대 설비를 예측하는 데 단일 서버로 10분 걸리던 것이 10노드 클러스터에서 1분으로 줄어듭니다.

​

Databricks 핵심 기능

​

운영 스펙

​

1. Champion 모델을 PySpark UDF로 로드

​

PySpark UDF란?

champion_udf = mlflow.pyfunc.spark_udf(
    spark,
    model_uri=f"models:/{model_name}@Champion",
    result_type="double"
)

주의 Spark UDF로 모델을 분산 실행하면, 각 노드에 모델이 복제됩니다. 모델 크기가 큰 경우(예: 딥러닝 모델 2GB 이상) 메모리 문제가 발생할 수 있습니다. XGBoost 같은 트리 모델은 보통 수 MB~수십 MB 수준이라 전혀 문제없지만, 비전 모델을 Spark UDF로 돌리려면 메모리 계획을 세워야 합니다. 이런 경우에는 Model Serving endpoint를 사용하는 것이 더 적합합니다.

​

2. 추론 데이터 준비

inference_df = (
    spark.table("lgit_pm_training")
    .select("udi", *feature_columns)
    .withColumn("inference_timestamp", F.current_timestamp())
)

​

3. 분산 배치 예측 + 위험 등급 부여

​

추론 결과에 포함되는 정보

​

위험 등급 분류 기준 및 현장 대응 방법

주의 위험 등급 임계값(0.3/0.5/0.8)은 교육용 예시입니다. 실무에서는 반드시 현장 엔지니어와 함께 결정해야 합니다. 처음에는 보수적으로(CRITICAL 기준을 낮게, 예: 0.7) 시작하고, 오탐이 많으면 점진적으로 기준을 올리는 것이 안전합니다. 현장의 신뢰를 한 번 잃으면 회복하는 데 6개월이 걸립니다.

preds_df = (
    inference_df
    .withColumn("failure_probability", champion_udf(*feature_columns))
    .withColumn("predicted_failure",
                F.when(F.col("failure_probability") > 0.5, 1).otherwise(0))
    .withColumn("risk_level",
        F.when(F.col("failure_probability") > 0.8, "CRITICAL")
        .when(F.col("failure_probability") > 0.5, "HIGH")
        .when(F.col("failure_probability") > 0.3, "MEDIUM")
        .otherwise("LOW"))
    .withColumn("model_name", F.lit(model_name))
    .withColumn("model_version", F.lit(int(champion_info.version)))
)

​

4. Delta Lake에 예측 결과 저장

​

왜 예측 결과를 저장해야 하는가?

1. 이력 추적 (Trending): “이 설비의 고장 확률이 지난 1주일 동안 어떻게 변했는가?” 분석 가능
2. 모델 모니터링: 예측 분포가 시간이 지남에 따라 변하는지 확인 (08번 노트북에서 활용)
3. 정비 계획 수립: 과거 예측 결과를 기반으로 설비별 정비 주기 최적화
4. 감사 추적 (Audit Trail): “언제, 어떤 모델이, 어떤 예측을 했는지” 완전한 기록 유지

preds_df.write.mode("append").option("mergeSchema", "true") \
    .saveAsTable("lgit_pm_inference_results")

참고 append vs overwrite는 사소해 보이지만 운영에서 매우 중요한 결정입니다. append를 써야 예측 이력이 쌓이고, 시간이 지나면서 “모델이 점점 CRITICAL을 많이 예측하고 있다 → 드리프트 의심” 같은 분석이 가능합니다. 예측 이력은 한 번 잃으면 복구할 수 없습니다. 디스크 비용은 싸지만, 3개월치 이력 데이터는 돈으로 살 수 없습니다.

​

Databricks UI 확인 포인트

1. Catalog > lgit_mlops_poc > Tables > lgit_pm_inference_results 클릭
2. Sample Data 탭: 예측 결과 (failure_probability, risk_level, 고장 유형 확률) 확인
3. Details 탭: 행 수가 이전보다 증가했는지 확인 (append 모드이므로 누적)
4. History 탭: 각 배치 실행의 타임스탬프 확인
5. 우측 상단 Create> Quick Dashboard: 위험 등급 분포 시각화를 즉시 생성 가능

​

5. 예측 결과 분석

-- 위험 등급별 분포
SELECT
  risk_level,
  COUNT(*) as count,
  ROUND(AVG(failure_probability), 4) as avg_failure_prob,
  ROUND(MIN(failure_probability), 4) as min_prob,
  ROUND(MAX(failure_probability), 4) as max_prob
FROM lgit_pm_inference_results
GROUP BY risk_level
ORDER BY avg_failure_prob DESC

-- CRITICAL/HIGH 위험 설비 목록 (즉시 점검 필요)
SELECT
  udi, product_quality, failure_probability, risk_level,
  air_temperature_k, rotational_speed_rpm, torque_nm, tool_wear_min,
  inference_timestamp
FROM lgit_pm_inference_results
WHERE risk_level IN ('CRITICAL', 'HIGH')
ORDER BY failure_probability DESC
LIMIT 20

성공 이 노트북은 Databricks Workflow에서 일 4회 자동 실행됩니다. model_version 컬럼이 함께 저장되므로, 모델 변경 전후의 예측 결과를 비교 분석할 수 있습니다.

​

6. 고장 유형별 확률 추정 (Fault Type Probability)

전체 노트북 코드: 06_batch_inference.py (고장 유형별 확률 추정 섹션)

# 학습 데이터의 고장 유형 비율 (AI4I 2020 기준)
fault_type_ratios = {
    "TWF": 0.10,   # Tool Wear Failure (공구 마모)
    "HDF": 0.35,   # Heat Dissipation Failure (열 방출 고장)
    "PWF": 0.25,   # Power Failure (전력 고장)
    "OSF": 0.20,   # Overstrain Failure (과부하 고장)
    "RNF": 0.10    # Random Failure (랜덤 고장)
}

# 고장 확률 × 유형별 비율 = 유형별 확률
for fault_type, ratio in fault_type_ratios.items():
    predictions = predictions.withColumn(
        f"prob_{fault_type.lower()}",
        F.round(F.col("failure_probability") * F.lit(ratio), 4)
    )

참고 이 방식은 사후 추정(post-hoc estimation)으로, 실제 운영에서는 고장 유형별 멀티레이블 분류 모델 을 별도로 학습하는 것을 권장합니다.

​

7. 우선순위 기반 유지보수 스케줄링

​

자동 알림 설정 방법

1. Databricks SQL Alert: CRITICAL 건수가 임계값을 초과하면 이메일/Slack 자동 알림
2. Workflows 알림: 배치 추론 Job 완료 시 결과 요약을 이메일로 발송
3. SQL Dashboard: 정비팀이 웹 브라우저에서 실시간으로 위험 설비 현황을 확인

​

8. 센서 해석 가이드

주의 예측 결과를 기존의 MES(Manufacturing Execution System)나 CMMS(Computerized Maintenance Management System)와 연동하면, 정비 작업 지시가 자동으로 생성되는 완전 자동화된 예지보전 시스템 을 구축할 수 있습니다.

​

9. Feature Store가 추론에서 빛나는 이유

전체 노트북 코드: 06_batch_inference.py (Feature Store 섹션)

​

Feature Store 없이 vs 있을 때

​

Feature Lookup 기반 추론 (참고 패턴)

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

fe = FeatureEngineeringClient()

lookups = [
    FeatureLookup(
        table_name="catalog.schema.lgit_pm_features",  # Feature Table
        feature_names=["temp_diff", "power_w", "strain", "risk_score"],
        lookup_key="udi"  # 조인 키 (설비 ID)
    )
]

# score_batch: 모델 + Feature Lookup을 한번에 실행
predictions = fe.score_batch(
    model_uri="models:/lgit_predictive_maintenance@Champion",
    df=new_sensor_data,  # 새로 들어온 센서 데이터 (udi만 있으면 됨!)
    feature_lookups=lookups
)

참고 new_sensor_data에는 udi(설비 ID)만 있으면 됩니다. temp_diff, power_w 등의 피처는 Feature Store가 자동으로 조인 합니다. 피처 계산 로직이 바뀌어도 추론 코드를 수정할 필요 없음— Feature Table만 업데이트하면 됩니다.

​

10. 모델이 변경되어도 추론이 끊기지 않는 이유

​

시나리오: Champion v5 → v6 교체 시 추론 코드 변화

​

만약 v6에 문제가 생기면? (롤백)

# 관리자가 한 줄 실행 — 즉시 v5로 롤백!
client.set_registered_model_alias("lgit_predictive_maintenance", "Champion", version=5)
# → 다음 추론부터 자동으로 v5 사용 (코드 수정 없음)

참고 이 구조 덕분에 “새 모델 배포”가 무서운 일이 아니게 됩니다. 잘못되면 30초 안에 롤백할 수 있으니까요. 이 안전장치가 없으면 팀은 모델 업데이트를 두려워하게 되고, 결국 6개월 된 낡은 모델로 운영하게 됩니다.

​

요약

​

이 노트북에서 수행한 작업

​

운영 이슈 3가지

1. 입력 테이블이 비어있는 경우— 업스트림 ETL이 실패하면 빈 테이블에 추론을 돌리게 됩니다. 반드시 입력 건수 체크 로직을 추가하세요.
2. 모델 버전 불일치— Champion 에일리어스가 실수로 삭제되면 에러가 납니다. 에일리어스 존재 여부를 먼저 확인하는 방어 코드가 필요합니다.
3. 디스크 용량 초과— append 모드로 수개월 쌓다 보면 테이블이 커집니다. 파티셔닝과 OPTIMIZE/VACUUM 전략을 미리 세워두세요.