MLPE 스터디 3주차 - Ch6.2 Evidently로 데이터 드리프트 모니터링하기

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=reference_data, current_data=current_data)
report.save_html("drift_report.html")

apiVersion: apps/v1
kind: Deployment
metadata: {name: evidently-ui, namespace: evidently}
spec:
  replicas: 1
  strategy: {type: Recreate}                       # PVC 가 RWO
  template:
    spec:
      nodeSelector: {kubernetes.io/arch: amd64}
      containers:
        - name: evidently-ui
          image: hyeonjae0913/evidently-ui:dd6b564
          args: ["--host", "0.0.0.0", "--port", "8000", "--workspace", "/app/workspace"]
          resources:
            requests: {cpu: 50m, memory: 256Mi}
            limits: {memory: 1Gi}
          volumeMounts: [{name: workspace, mountPath: /app/workspace}]
      volumes:
        - name: workspace
          persistentVolumeClaim: {claimName: evidently-workspace}   # local-path 1Gi
---
apiVersion: v1
kind: Service
metadata:
  name: evidently-ui
  namespace: evidently
  annotations: {lbipam.cilium.io/ips: "192.168.0.245"}   # 클러스터 안: evidently-ui.evidently.svc:8000
spec:
  type: LoadBalancer
  selector: {app: evidently-ui}
  ports:
    - {name: http, port: 8000, targetPort: http}
    - {name: http-80, port: 80, targetPort: http}

from evidently.ui.workspace import RemoteWorkspace

ws = RemoteWorkspace("http://192.168.0.245")
project = ws.create_project(
    name="income-classifier",
    description="Used to classify users into multiple income bands",
)

DashboardPanelCounter(
    filter=ReportFilter(metadata_values={}, tag_values=[]),   # 어떤 스냅숏을 쓸지. 비우면 전부
    agg=CounterAgg.NONE,                                      # 숫자를 계산하지 않는다
    title="Income Classifier Batch Data Drift Dashboard",
)
DashboardPanelCounter(
    title="Share of Drifted Features",
    filter=ReportFilter(metadata_values={}, tag_values=[]),
    value=PanelValue(
        metric_id="DatasetDriftMetric",           # 어느 메트릭의 결과에서
        field_path="share_of_drifted_columns",    # 어느 필드를 꺼낼지
        legend="share",
    ),
    text="share",
    agg=CounterAgg.LAST,      # 스냅숏이 여러 개 쌓여도 가장 최근 값 하나
    size=1,                   # 1 은 절반 너비, 2 는 전체 너비
)

def detect_drift(model_name, model_stage, mlflow_host, minio_host, access_key, secret_key,
                 reference_dataset_name, feature_dataset_path,
                 evidently_workspace_url, evidently_ui_project_name):
    # 1) Production 모델의 run 에서 reference 데이터셋 위치를 찾는다
    for dataset_input in mlflow.get_run(model_run_id).inputs.dataset_inputs:
        for tag in dataset_input.tags:
            if tag.value == reference_dataset_name:
                dataset_source = json.loads(dataset_input.dataset.source)["uri"]
    # 2) MinIO(홈랩은 RustFS)에서 reference 를, 앞 단계 출력에서 current 를 읽는다
    reference_df = pd.read_csv(file_path)
    feature_df = pd.read_parquet(feature_dataset_path).drop(columns=["user_id"], errors="ignore")
    # 3) 대시보드 패널이 읽는 메트릭과 같은 것으로 리포트를 만든다
    report = Report(metrics=[DatasetDriftMetric(), DatasetMissingValuesMetric(),
                             ColumnDriftMetric(column_name="Education"), ColumnSummaryMetric(column_name="Education"),
                             ...])   # Marital-Status, Native_country, Occupation, Race, Relationship, Sex, Workclass
    report.run(reference_data=reference_df, current_data=feature_df)
    # 4) 프로젝트에 스냅숏으로 올린다
    evidently_workspace.add_report(get_evidently_project(evidently_workspace, evidently_ui_project_name).id, report)

name: detect_drift
inputs:
  - {name: model_name, type: String}
  - {name: model_stage, type: String}
  - {name: mlflow_host, type: String}
  - {name: minio_host, type: String}
  - {name: reference_dataset_name, type: String}
  - {name: feature_dataset_path, type: Artifact}
  - {name: evidently_workspace_url, type: String}
  - {name: evidently_ui_project_name, type: String}
implementation:
  container:
    image: hyeonjae0913/detect-drift:dd6b564
    command:
      - sh
      - -c
      # 키는 Secret env 로 받는다
      - exec python3 /app/src/detect_drift/detect_drift.py "$@" --access_key "$AWS_ACCESS_KEY_ID" --secret_key "$AWS_SECRET_ACCESS_KEY"
      - sh
      - --model_name
      - {inputValue: model_name}
      # ... model_stage, mlflow_host, minio_host, reference_dataset_name
      - --feature_dataset_path
      - {inputPath: feature_dataset_path}
      # ... evidently_workspace_url, evidently_ui_project_name

@dsl.pipeline(name="income-classifier-pipeline-with-drift")
def income_classifier_pipeline_with_drift(
    minio_host: str = "rustfs.mlflow.svc:9000",
    mlflow_host: str = "http://mlflow.mlflow.svc:5000",
    reference_dataset_name: str = "reference_features",
    evidently_workspace_url: str = "http://evidently-ui.evidently.svc:8000",   # Pod 안에서 쓰이므로 클러스터 주소
    evidently_ui_project_name: str = "income-classifier",
    output_file_name: str = "inference_result_with_drift.parquet", ...):
    fetch = homelab(read_op(...))
    feats = homelab(features_op(..., entity_df=fetch.outputs["data_output"]))
    drift = homelab(drift_op(..., feature_dataset_path=feats.outputs["data_output"]))
    infer = homelab(infer_op(..., input_data=feats.outputs["data_output"])).after(drift)
    homelab(write_op(..., input_data=infer.outputs["data_output"]))

def homelab(task):        # Secret rustfs-credentials 를 env 로, 캐시 끔 (5장과 같다)
    kubernetes.use_secret_as_env(task, secret_name="rustfs-credentials",
                                 secret_key_to_env={"AWS_ACCESS_KEY_ID": "AWS_ACCESS_KEY_ID",
                                                    "AWS_SECRET_ACCESS_KEY": "AWS_SECRET_ACCESS_KEY"})
    task.set_caching_options(False)
    return task

class MonitoringService:
    def iterate(self, new_rows: pandas.DataFrame):
        self.current = pandas.concat([self.current, new_rows], ignore_index=True)   # 새 요청을 윈도 뒤에
        current_size = self.current.shape[0]
        if current_size > self.window_size:                                         # 최근 N 개만 남긴다
            self.current = self.current.iloc[-self.window_size:].reset_index(drop=True)
        if current_size < self.window_size:                                         # 아직 덜 찼으면 기다린다
            logger.info(f"Not enough data for measurement: {current_size} of {self.window_size}. Waiting more data")
            return
        logger.info("Running report")
        self.report.run(reference_data=self.reference, current_data=self.current)
        self.workspace.add_report(project_id=self.project_id, report=self.report)

service: "service_with_drift_detection:svc"
labels: {owner: ml-engineering-team, env: homelab, variant: drift}
include:
  - "service_with_drift_detection.py"
  - "homelab.env"
  - "src/*.py"
  - "data_drift.py"
  - "evidently_reports.py"
  - "requirements_homelab_drift.txt"
python:
  requirements_txt: requirements_homelab_drift.txt
docker:
  base_image: python:3.12-slim
  env:
    - ENV_NAME=homelab

apiVersion: serving.yatai.ai/v2alpha1
kind: BentoDeployment
metadata: {name: income-classifier-drift, namespace: yatai}
spec:
  bento: income-classifier-drift-l2mj57gz42lndbch
  ingress: {enabled: false}
  autoscaling: {minReplicas: 1, maxReplicas: 1}      # 복제본마다 윈도가 따로 쌓인다
  envs:
    - {name: ENV_NAME, value: homelab}
    - {name: BENTOML_CONFIG_OPTIONS, value: "api_server.workers=1"}   # 워커마다 윈도가 따로 쌓인다
    - {name: MLFLOW_S3_ENDPOINT_URL, value: "http://rustfs.mlflow.svc:9000"}
    - {name: K8S_SECRET_NAME, value: income-classifier-env}
    - {name: EVIDENTLY_PROJECT_ID, value: "01a1b111-77ef-7c70-8281-b55822c671c6"}
  extraPodSpec:
    nodeSelector: {kubernetes.io/arch: amd64}
    serviceAccountName: income-classifier
  runners:
    - name: random-forest-classifier
      autoscaling: {minReplicas: 1, maxReplicas: 1}
# + LB Service income-classifier-drift-lb (192.168.0.246)