Lineage Emit 디자인 (#1281 Gap 3 M3)

pipelines/gend_pipelines/ml/batch_predict_asset.py 의 success path 에서 POST /api/v1/lineage/edges 자동 호출 — MLflow 모델 → Iceberg Gold 테이블 lineage 엣지를 ArangoDB 에 자동 등록.

디자인 결정

1. 왜 asset 종료 후 emit?

batch_predict 의 input/output 이 명확: input=mlflow.{alias}.{model}, output=iceberg.gold.{table}
Dagster @asset 의 MaterializeResult 반환 직전이 emit 의 자연스러운 hook 지점
success path 만 (rows_in == 0 early-return 시 emit 하지 않음 — 빈 적재는 lineage 의미 없음)

2. 왜 fire-and-forget?

lineage 는 부가 메타데이터. 실패 시 asset materialization 실패시키지 않음
gend-api 일시 장애 시 batch_predict 결과는 정상 적재되어야 함
5초 timeout + context.log.warning 만 (Dagster step 실패 안 함)

3. 왜 가상 catalog 형식?

ArangoDB data_lineage edge collection 은 catalog.schema.table 3-segment 표준 사용
MLflow 모델은 catalog/schema 가 없음 → 가상 형식: mlflow.{alias}.{model_name}
예: mlflow.Production.credit_risk → iceberg.gold.predictions_demo_credit_risk
Lineage UI 가 그래프 시각화 시 일관 형식으로 처리

코드 깊이

# pipelines/gend_pipelines/ml/lineage_emit.py
def emit_batch_predict_lineage(
    model_name: str,
    model_alias: str,
    model_version: str,
    output_table: str,
    ml_batch_run_id: str,
    dagster_job_name: str,
    *,
    gend_api_url: str | None = None,
    internal_token: str | None = None,
) -> bool:
    """Fire-and-forget lineage emit. Returns True on 2xx, False otherwise."""
    api_url = gend_api_url or os.getenv("GEND_API_URL", "...")
    token = internal_token or os.getenv("GEND_INTERNAL_TOKEN", "")
    payload = {
        "source_table": f"mlflow.{model_alias}.{model_name}",
        "target_table": output_table,
        "lineage_type": "ml_batch_predict",
        "pipeline": f"dagster:{dagster_job_name}:{ml_batch_run_id}",
    }
    try:
        resp = requests.post(
            f"{api_url.rstrip('/')}/api/v1/lineage/edges",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            json=payload,
            timeout=5,
        )
        return 200 <= resp.status_code < 300
    except Exception as exc:
        logger.warning("lineage emit failed (silent): %s", exc)
        return False

호출 다이어그램

회귀 가드

pipelines/tests/test_ml_batch_lineage_emit.py:

200 응답 → True
500/Timeout → False (no exception)
env override (GEND_API_URL, GEND_INTERNAL_TOKEN)
payload 형식 (source/target/type/pipeline)
asset 통합: lineage_emitted: bool metadata

트러블슈팅

증상	원인	해결
ArangoDB 에 edge 안 보임	gend-api 도달 못함	`GEND_API_URL` env 확인 + cluster DNS
401 인증 실패	`GEND_INTERNAL_TOKEN` 미설정	Vault `secret/gend/dagster-internal-token` 참조
Asset SUCCESS 인데 lineage 없음	fire-and-forget — 정상 silent 동작	Dagster log 의 `lineage emit failed (silent)` warning 확인

디자인 결정​

1. 왜 asset 종료 후 emit?​

2. 왜 fire-and-forget?​

3. 왜 가상 catalog 형식?​

코드 깊이​

호출 다이어그램​

회귀 가드​

트러블슈팅​

관련​