본문으로 건너뛰기

M3 E2E 실증 — batch_predict full lineage (#1430)

#1356 sink VARCHAR coercion fix 가 prod 에서 작동함을 확인하는 end-to-end 실증.

사전 시드 (Step B)​

MLflow demo_credit_risk@Production v2​

scripts/seed_demo_credit_risk.py (#1358 / PR #1398) 를 Dagster daemon pod 에서 실행:

DAG_POD=$(kubectl --context aks-genos-prod -n gend get pod -l app=dagster-daemon -o jsonpath='{.items[0].metadata.name}')
kubectl cp scripts/seed_demo_credit_risk.py $DAG_POD:/tmp/ -c dagster-daemon
kubectl exec $DAG_POD -c dagster-daemon -- python /tmp/seed_demo_credit_risk.py --skip-trino

산출물 — MLflow run 7b243200bba4418497e8a5c173a94868, score 0.862, registered as demo_credit_risk@Production v2.

Trino iceberg.silver.demo_customers​

kubectl exec $DAG_POD -c dagster-daemon -- bash -c '
  export TRINO_HOST=trino.gend.svc.cluster.local
  export TRINO_PORT=8080
  python /tmp/seed_demo_credit_risk.py --skip-mlflow
'

#1430 운영 회귀 발견 1 — K8s 가 자동 주입하는 TRINO_PORT=tcp://<ip>:8080 환경변수가 스크립트의 os.getenv("TRINO_PORT", "8080") 와 충돌. 호출 측에서 TRINO_PORT=8080 명시 override 필요. follow-up: 스크립트가 K8s service env 패턴을 감지해 정리하거나 변수명 prefix 변경.

iceberg.silver.demo_customers 10 rows 확인.

ml_batch_job 등록​

직접 SQL 로 row insert (worktree-ship #1430 의 일회성 실증 — kind dev 의 scripts/seed_ml_batch_demo.py 는 prod 의 GEND_* settings 검증 통과 못해 사용 불가):

INSERT INTO ml_batch_job (id, workspace_id, name, model_name, model_alias, source_asset,
  feature_columns, id_columns, output_table, executor, drift_check, drift_threshold,
  status, owner_id, created_at, updated_at)
VALUES (gen_random_uuid(), 'a841b36e-9ba4-4c74-bb42-761058902bf3',
  'demo_credit_risk', 'demo_credit_risk', 'Production',
  'iceberg.silver.demo_customers',
  '["age", "income", "credit_history_len"]'::jsonb,
  '["customer_id"]'::jsonb,
  'iceberg.gold.demo_credit_risk_predictions',
  'pandas', false, 0.3, 'active',
  'ab1b0d53-6ea9-47df-8145-e09aa8595f56', NOW(), NOW());

ml_batch_job id 196e9c10-632f-479a-8fe6-0e885c7c15e6 등록.

실행 (Step D)​

Dagster GraphQL launchRun​

mutation LaunchRun {
  launchRun(executionParams: {
    selector: {
      repositoryLocationName: "gend_pipelines:defs"
      repositoryName: "__repository__"
      jobName: "ml_batch_demo_credit_risk_job"
    }
    runConfigData: "{}"
    mode: "default"
  }) { ... on LaunchRunSuccess { run { runId status } } }
}

첫 실행 (172d33fc-8837-4ae2-8fd8-cfb404b81241): FAILURE — TrinoUserError COLUMN_NOT_FOUND ml_batch_run_id.

#1430 운영 회귀 발견 2 — 옛 schema (mlflow_run_id, model_version, predicted_at, 4 features) 의 iceberg.gold.predictions_demo_credit_risk 가 10000 rows 로 prod 에 잔존. sink 의 CREATE TABLE IF NOT EXISTS 는 no-op, 그러나 INSERT 가 새 schema (ml_batch_run_id) 컬럼을 시도 → fail.

운영 처리: DROP TABLE IF EXISTS iceberg.gold.predictions_demo_credit_risk 후 재실행.

회귀 가드 후보: sink 의 _validate_table_schema_matches_sink — CREATE TABLE 직후 SELECT column 이 sink 의 INSERT 컬럼 집합과 정확 일치하는지 검증 + 명확한 ERROR. 별도 issue 발행 권장.

재실행 결과​

두번째 실행 58c65042-7c59-47d5-a91c-c2e0426b1406: SUCCESS.

DESCRIBE iceberg.gold.predictions_demo_credit_risk
  customer_id        varchar          ← #1356 VARCHAR coercion 작동 (id_columns BIGINT → varchar)
  prediction         double
  ml_batch_run_id    varchar
  predicted_at       timestamp(6)

SELECT COUNT(*) → 10
SELECT * LIMIT 3
  ['1', 1.0, '58c65042-7c59-47d5-a91c-c2e0426b1406', 2026-05-30 09:35:44]
  ['2', 1.0, '58c65042-7c59-47d5-a91c-c2e0426b1406', 2026-05-30 09:35:44]
  ['3', 1.0, '58c65042-7c59-47d5-a91c-c2e0426b1406', 2026-05-30 09:35:44]

#1356 fix 작동 검증 evidence:

  • id_columns=["customer_id"] 가 source 에서 BIGINT 였으나 sink CREATE TABLE 가 VARCHAR 로 정의 → INSERT 가 BIGINT 값을 VARCHAR 로 코어션 (_sql_id_literal('1') 결과 '1').
  • ml_batch_run_id 컬럼 = Dagster run id (lineage trace).

Lineage 결과​

LayerID비고
MLflow run7b243200bba4418497e8a5c173a94868seed (model train)
MLflow modeldemo_credit_risk@Production v2sklearn LogisticRegression score 0.862
Dagster run58c65042-7c59-47d5-a91c-c2e0426b1406ml_batch_demo_credit_risk_job
Iceberg goldiceberg.gold.predictions_demo_credit_risk10 rows, ml_batch_run_id = dagster run_id

ml_batch_run PG row 누락 — SELECT * FROM ml_batch_run ORDER BY created_at DESC LIMIT 3 결과 0 rows.

#1430 운영 회귀 발견 3 — ml_batch_demo_credit_risk_job (demo asset) 는 PG ml_batch_run 테이블에 row 를 안 만듦. production batch_predict job (factory-minted) 만 ml_batch_run insert. demo asset 의 lineage 가 Dagster + Iceberg 까지만 도달하고 GenD UI (/ml-batch/runs) 에 안 나타남. follow-up: demo asset 도 ml_batch_run row insert 하거나, 별도 ml_batch_job 으로 factory-minted asset 사용.

인프라 수정 사항​

(본 PR 산출물에 포함 — 본 절차 재현에 필요)

  • infra/dagster/configmap.yaml — workspace.yaml 에 ml_plugin_train_repo location 추가 (시나리오 1 차단 해소).
  • apps/api/src/gend_api/services/ml_plugins/dagster_trigger.py — selector 의 repositoryName 을 __repository__ 로 fix (location_name 과 repository_name 분리).