ML Batch Inference

GenD 의 ML Batch Inference 는 MLflow Registry 에 등록된 모델을 Dagster asset 으로 자동 호출하여 Iceberg Gold 테이블에 점수를 적재합니다. 야간 신용평가·이상탐지·수요예측 같은 정기 배치 추론 시나리오를 표준화합니다.

본 가이드는 M2 단계 (Epic #1083) 에서 사용 가능한 기능을 다룹니다. M2 는 M1 의 read-only API 위에 Mutation API + manual trigger + drift_check 활성 을 추가합니다. 다음 마일스톤(M3 실 Dagster GraphQL trigger + lineage emit + DataMart 자동 등록 + UI) 은 후속 가이드에서 다룹니다.

핵심 개념

개념	정의
Batch Job (`ml_batch_job`)	모델 + 입력 feature asset + 출력 Gold 테이블 + 스케줄 의 결합 단위. 워크스페이스 격리
Batch Run (`ml_batch_run`)	Job 의 단일 실행 인스턴스. Dagster run + MLflow run + rows in/out + 드리프트 점수 기록
Model alias	MLflow Model Registry 의 alias-only 정책 — `models:/<name>@Production` URI 사용. Stage 기반 (`Production`/`Staging`) 사용 금지 (회귀 가드 단언)
Executor	M1 은 `pandas` 만 (단일 노드 Python + `mlflow.pyfunc`). M3 에서 `spark` 옵션 추가
Gold sink	Iceberg 테이블 + `extra_properties['gend.layer']='gold'` 자동 부착, DELETE→INSERT 재적재 패턴

데이터 모델

`ml_batch_job` (18 columns)

class MLBatchJob(Base):
    id: UUID                    # PK
    workspace_id: UUID          # FK → workspace.id (M1 nullable, M2 backfill 후 NOT NULL)
    name: str                   # UNIQUE per workspace
    description: str | None
    model_name: str             # MLflow registered model
    model_alias: str            # default 'Production' (alias-only 정책)
    source_asset: str           # Dagster AssetKey (점 구분, e.g. "iceberg.silver.demo_customers")
    feature_columns: list[str]  # JSONB
    id_columns: list[str]       # JSONB
    output_table: str           # iceberg.gold.<table>
    schedule_cron: str | None   # NULL = manual only
    executor: str               # 'pandas' (M1 강제), 'spark' (M3)
    drift_check: bool           # M1 False 강제
    drift_threshold: float      # default 0.3 (M2 활성)
    status: str                 # active | paused | deleted
    owner_id: UUID              # plain UUID (Keycloak user id 사용, FK 아님)
    created_at, updated_at

`ml_batch_run` (15 columns)

class MLBatchRun(Base):
    id: UUID                    # PK
    job_id: UUID                # FK CASCADE → ml_batch_job.id
    dagster_run_id: str | None  # Dagster run UUID
    mlflow_run_id: str | None   # MLflow run ID
    model_version: str | None   # alias 해결 후 실제 점수화에 쓰인 버전 (run 시작 시 1회 fix, hot-swap 방지)
    status: str                 # queued | running | succeeded | failed | cancelled
    started_at, finished_at
    rows_in: int | None
    rows_written: int | None
    drift_score: float | None   # (M2 활성)
    drift_status: str | None    # ok | warn | breach (M2 활성)
    error_message: str | None
    artifact_uri: str | None    # s3://.../drift_report.html (M2)
    created_at

Index: idx_ml_batch_run_job_started(job_id, started_at DESC) + partial idx_ml_batch_run_status WHERE status IN ('queued','running') (PostgreSQL 전용, _PG_POST_MIGRATIONS).

REST API

Base: /api/v1/ml/batch, JWT 보호, 워크스페이스 격리 (caller tenant_slug → Workspace.id)

Surface	Endpoint	Role	Milestone
Read	`GET /jobs`, `GET /jobs/{id}`, `GET /runs`, `GET /runs/{id}`	viewer	M1
Mutation — create	`POST /jobs`	admin	M2
Mutation — update	`PUT /jobs/{id}`	admin	M2
Mutation — soft delete	`DELETE /jobs/{id}`	admin	M2
Manual trigger	`POST /jobs/{id}/run`	admin	M2 (row 생성만, Dagster launch 는 M3)

Read-only (M1)

`GET /jobs`

GET /api/v1/ml/batch/jobs?limit=20&offset=0 HTTP/1.1
Authorization: Bearer <JWT>

응답 (200):

{
  "items": [
    {
      "id": "...",
      "name": "demo_credit_risk",
      "model_name": "credit_risk",
      "model_alias": "Production",
      "source_asset": "iceberg.silver.demo_customers",
      "output_table": "iceberg.gold.predictions_demo_credit_risk",
      "executor": "pandas",
      "schedule_cron": "0 1 * * *",
      "status": "active"
    }
  ],
  "total": 1,
  "limit": 20,
  "offset": 0
}

`GET /jobs/{id}`

200: 단일 job
403: cross-workspace 접근
404: 미존재 또는 다른 workspace
422: 잘못된 UUID

`GET /runs`

쿼리 파라미터: job_id (UUID, optional), status (optional), limit (default 50), offset.

GET /api/v1/ml/batch/runs?job_id=<uuid>&status=succeeded&limit=10 HTTP/1.1

`GET /runs/{id}`

200: 단일 run + parent job 의 workspace 검증 통과
403/404/422: 동일 패턴

Mutation API (M2)

모든 mutation 은 admin 전용 (require_admin). M3 에서 POST /jobs/{id}/run 은 viewer-with-ABAC ModelGrant 로 완화 예정 (#1228 통합).

`POST /jobs` — 새 job 정의

curl -X POST https://gend.genon.ai/api/v1/ml/batch/jobs \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "demo_credit_risk",
    "description": "신용평가 야간 배치",
    "model_name": "credit_risk",
    "model_alias": "Production",
    "source_asset": "iceberg.silver.demo_customers",
    "feature_columns": ["age", "income", "debt_ratio"],
    "id_columns": ["customer_id"],
    "output_table": "iceberg.gold.predictions_demo_credit_risk",
    "schedule_cron": "0 1 * * *",
    "executor": "pandas",
    "drift_check": true,
    "drift_threshold": 0.3
  }'

응답: 201 + MLBatchJobRead (서버가 workspace_id 와 owner_id 자동 스탬프).

응답 코드:

201: 성공
403: admin 아님 (require_admin 거부) 또는 cross-workspace 차단
409: (workspace_id, name) UNIQUE 위반 (이미 존재)
422: Pydantic 검증 실패 (필수 필드 누락, feature_columns=[], executor='spark', drift_threshold > 1.0 등)

드리프트 정책 (M2 — Drift G-1 통합): drift_check 기본값이 True 로 변경. 신규 job 은 기본 drift 모니터링 활성. 본 PR 은 asset factory 가드 완화 + 인터페이스 stub 까지; 실제 Iceberg fetch + Evidently HTML 업로드 + DriftService.compute_and_save 호출은 M3 에서 Dagster webhook 과 함께 wire-up.

`PUT /jobs/{id}` — 변경 (mutable 필드 한정)

curl -X PUT https://gend.genon.ai/api/v1/ml/batch/jobs/$ID \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "description": "신규 노트",
    "schedule_cron": "0 2 * * *",
    "drift_check": true,
    "drift_threshold": 0.5,
    "status": "paused"
  }'

Mutable: description / schedule_cron / drift_check / drift_threshold / status (active|paused 만).

Immutable (PUT 시 422):

필드	사유
`name` / `workspace_id`	UNIQUE 합성키. 이름 변경 = 다른 job → 새 POST.
`model_name` / `model_alias`	라이브 스케줄의 점수화 로직이 silent 하게 swap 될 위험 — 감사 추적 필수.
`source_asset` / `feature_columns` / `id_columns`	같은 스케줄이 전혀 다른 행/형태를 점수화하게 됨 — 새 POST 강제.
`output_table`	싱크 이동은 downstream DataMart consumer 를 고아로 만듦.
`executor` / `owner_id`	M2 는 pandas 만. owner 변경은 별도 audit 이벤트 필요.

Status whitelist: active / paused 만. deleted 는 DELETE 전용 (silent stealth-delete 차단).

Pydantic extra='forbid': 위 immutable 필드 또는 미지정 키 전송 시 422 (라우터 도달 전 차단). 서비스 레이어 second-guard 도 422 매핑 (Dagster/스크립트 비-라우터 호출자 보호).

`DELETE /jobs/{id}` — soft delete

curl -X DELETE https://gend.genon.ai/api/v1/ml/batch/jobs/$ID \
  -H "Authorization: Bearer $JWT"

204: status='deleted' 로 flip (행은 보존). MLBatchRun 이력 (FK CASCADE 미발생, status flip 만) 도 그대로 유지.
404: 미존재
403: admin 아님 또는 cross-workspace

Idempotent: 이미 deleted 인 job 을 다시 DELETE 해도 204 (audit emit 동반).

M3 영역: deleted_at / deleted_by 컬럼 추가 (현재 ADR-002 baseline 정책 상 Alembic migration 금지 → 컬럼 변경 보류).

`POST /jobs/{id}/run` — 수동 실행

curl -X POST https://gend.genon.ai/api/v1/ml/batch/jobs/$ID/run \
  -H "Authorization: Bearer $JWT"

201: 새 MLBatchRun row (status=queued, started_at=NULL) 생성 + 응답
404: parent job 미존재
409: parent job 이 soft-deleted (status='deleted')
403: admin 아님 또는 cross-workspace

중요 — M2 영역 제한: 본 PR 은 MLBatchRun 행 생성까지만. 실 Dagster GraphQL launchRun 호출은 M3. 운영자는 Dagster UI 에서 queued 행을 보고 수동으로 launch 하거나, sensor 가 queued → running 으로 promote.

Audit emit

모든 mutation 은 gend.audit logger 에 구조화 라인 emit:

Action	필드
`ml_batch.job.create`	`job_id`, `workspace`, `name`, `model`, `source`, `output`, `drift_check`, `user`
`ml_batch.job.update`	`job_id`, `name`, `changed`, `previous`, `user`
`ml_batch.job.soft_delete`	`job_id`, `name`, `model`, `owner`, `previous_status`, `deleted_by`
`ml_batch.job.run`	`job_id`, `run_id`, `name`, `owner`, `triggered_by`

AKS prod 에서 grep:

kubectl logs -n gend -l app=gend-api | grep "ml_batch\."

M2 는 logger emit 까지. M3 에서 audit_chain HMAC 파이프라인과 통합 (#520 P14 L2 위에 wire-up).

Dagster Asset Factory

pipelines/gend_pipelines/ml/batch_predict_asset.py:

from pipelines.gend_pipelines.ml.batch_predict_asset import make_batch_predict_asset

CREDIT_RISK_DAILY = make_batch_predict_asset(
    job_name="demo_credit_risk",
    model_name="credit_risk",                        # MLflow registered model name
    model_alias="Production",                        # alias-only (Stage 사용 금지)
    feature_asset=AssetKey(["iceberg", "silver", "demo_customers"]),
    feature_columns=["age", "income", "debt_ratio"],
    id_columns=["customer_id"],
    output_table="iceberg.gold.predictions_demo_credit_risk",
    schedule_cron="0 1 * * *",                       # 01:00 UTC 일배치
    executor="pandas",                               # M1: 강제 (M3 spark)
    drift_check=True,                                # M2: True 허용 (M1 강제 False 가드 완화)
)

M2 변경 (Drift G-1 통합): 기존 drift_check=True → ValueError 가드가 완화되어 True 를 수용. asset 은 MaterializeResult metadata 에 drift flag 를 emit; 실제 Iceberg historical fetch + Evidently HTML 업로드 + DriftService.compute_and_save 호출은 M3 에서 Dagster webhook 으로 wire-up.

반환값: (AssetsDefinition, ScheduleDefinition) 튜플. definitions.py 가 자동 등록.

Asset 실행 순서

upstream materialize — iceberg.silver.demo_customers (Trino 로 SELECT)
load model — MlflowResource 가 models:/credit_risk@Production URI 로 pyfunc 모델 로드 + signature 검증 (mismatch 시 fail-fast)
predict — pandas DataFrame 입력 → model.predict(...) → score 컬럼 추가
sink — Iceberg Gold 테이블에 DELETE→INSERT 재적재 + gend.layer='gold' Iceberg extra_properties 부착
emit metadata — Dagster MaterializeResult 에 rows in/out, MLflow run id, model version (alias 해결 후 fixed)

모델 alias 정책 (alias-only)

MLflow 2.9+ 에서 Stage (Production/Staging) 가 deprecated. 본 Epic 은 alias-only 정책으로 고정:

✅ models:/<name>@<alias> URI (예: models:/credit_risk@Production)
✕ mlflow.MlflowClient.transition_model_version_stage(stage="Production", ...)
✕ 코드 안에 Stage="Production" / stage='Production' 하드코딩

회귀 가드: apps/api/tests/test_no_mlflow_stage_hardcoding.py 가 apps/api/src/gend_api + pipelines/gend_pipelines 코드를 grep, 위반 0건 단언. 별도 sub-디렉토리 추가 시 scope 확장 필요.

Alias 이동 = 사실상 프로덕션 배포

운영팀이 MLflow Registry 에서 alias 를 옮기는 순간이 사실상 모델 프로덕션 배포입니다. M2 에서:

alias 변경 감지 webhook → audit emit + Slack 알림
ABAC + approval workflow (services/approval_service 재사용)

시드 데이터

데모 환경 부트스트랩:

cd apps/api && .venv/bin/python ../../scripts/seed_ml_batch_demo.py

생성되는 데이터:

demo_credit_risk job 1건 (active, 01:00 UTC cron)
5 dummy runs (status mix: succeeded × 3, failed × 1, running × 1)

스크립트는 idempotent — (workspace_id, name) UNIQUE 제약 위반 시 skip.

테스트

Suite	위치	검증
API router	`apps/api/tests/test_ml_batch_router.py` (38 tests)	GET 4종 + POST/PUT/DELETE/run 4종, admin 강제, workspace 격리·스탬핑, 409 UNIQUE, immutable 422, soft delete, viewer 403, 404/422
Registry	`apps/api/tests/test_ml_batch_registry.py` (22 tests)	CRUD, pagination, filter, create_job/update_job/soft_delete_job/trigger_run, immutable guard, deleted job trigger 409
Regression — alias-only	`apps/api/tests/test_no_mlflow_stage_hardcoding.py`	`Stage="Production"` 하드코딩 0건
Regression — protected routers	`apps/api/tests/test_protected_routers_registration.py`	ml_batch 라우터 JWT 의존성 등록
Dagster asset	`pipelines/tests/test_ml_batch_predict_asset.py` (15 tests)	factory, signature fail-fast, sink DELETE→INSERT, M2: drift_check=True 수용

전체 실행:

cd apps/api && .venv/bin/python -m pytest tests/test_ml_batch_*.py -v
cd pipelines && .venv/bin/python -m pytest tests/test_ml_batch_predict_asset.py -v

다음 마일스톤

Milestone	범위	상태
M1 (PoC)	read-only API + Dagster asset factory + alias-only 정책 + 4 GET endpoints	✓ (PR #1095)
M2 (Beta, partial)	Mutation API (POST/PUT/DELETE/run) + admin-only + workspace stamping + drift_check 활성 (interface stub) + audit emit + 60+ tests	✓ (본 PR)
M3 (GA)	실 Dagster GraphQL `launchRun` + lineage emit + DataMart 자동 등록 + Evidently HTML 업로드 + UI `components/ml-batch/` + `deleted_at` 컬럼 + ABAC ModelGrant 통합 + `executor='spark'` + Feast online sync + Slack/Email 알림 + Prometheus 메트릭 4종 + audit_chain HMAC 통합	(미착수)

M3 명시적 영역 (본 PR 미포함)

실 Dagster GraphQL launchRun 호출 — 현재 POST /jobs/{id}/run 은 ml_batch_run 행을 status=queued 로만 insert. Dagster sensor 또는 운영자 promote 가 필요.
Lineage emit — OpenLineage 이벤트 (source asset → model run → output table).
DataMart 자동 등록 — output_table 을 DataMart Gold 레이어로 자동 등록.
Drift 실 호출 — DriftService.compute_and_save 의 Iceberg historical fetch + Evidently HTML S3 업로드. 본 PR 은 interface stub (asset factory 가드만 완화).
UI — ui/src/components/ml-batch/ 페이지 (job list / detail / new job form / run history).
Soft delete tombstone — deleted_at / deleted_by 컬럼 (ADR-002 baseline-only 정책 해제 후 Alembic migration).
ABAC integration — POST /jobs/{id}/run 을 ModelGrant (action='predict') 위반 시 403 으로 제한 (현재는 admin only).

핵심 개념​

데이터 모델​

ml_batch_job (18 columns)​

ml_batch_run (15 columns)​

REST API​

Read-only (M1)​

GET /jobs​

GET /jobs/{id}​

GET /runs​

GET /runs/{id}​

Mutation API (M2)​

POST /jobs — 새 job 정의​

PUT /jobs/{id} — 변경 (mutable 필드 한정)​

DELETE /jobs/{id} — soft delete​

POST /jobs/{id}/run — 수동 실행​

Audit emit​

Dagster Asset Factory​

Asset 실행 순서​

모델 alias 정책 (alias-only)​

Alias 이동 = 사실상 프로덕션 배포​

시드 데이터​

테스트​

다음 마일스톤​

M3 명시적 영역 (본 PR 미포함)​

관련 문서 / 이슈​