如何构建端到端机器学习血缘（Lineage）：DVC、Evently AI 与 Prefect 实践指南

$dvc init

.
.dvc/
│
└── cache/ # [.gitignore] 存储dvc缓存（缓存实际数据文件）
└── tmp/ # [.gitignore]
└── .gitignore # gitignore 缓存, tmp, 和 config.local
└── config # 生产环境的dvc配置
└── config.local # [.gitignore] 本地dvc配置

stages: etl_pipeline: # dvc在此阶段将运行的主要命令 cmd: python src/data_handling/etl_pipeline.py # 运行主命令所必需的依赖项 deps: - src/data_handling/etl_pipeline.py - src/data_handling/ - src/_utils/ # dvc跟踪的输出路径 outs: - data/original_df.parquet - data/processed_df.parquet

import os
import argparse
import src.data_handling.scripts as scripts
from src._utils import main_logger

def etl_pipeline():
    # 提取所有数据
    df = scripts.extract_original_dataframe()
    
    # 保存原始parquet文件
    ORIGINAL_DF_PATH = os.path.join('data', 'original_df.parquet')
    df.to_parquet(ORIGINAL_DF_PATH, index=False)
    # dvc 跟踪
    
    # 转换
    df = scripts.structure_missing_values(df=df)
    df = scripts.handle_feature_engineering(df=df)
    
    PROCESSED_DF_PATH = os.path.join('data', 'processed_df.parquet')
    df.to_parquet(PROCESSED_DF_PATH, index=False)
    # dvc 跟踪
    
    return df

# 用于dvc执行
if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="run etl pipeline")
    parser.add_argument('--stockcode', type=str, default='', help="specific stockcode to process. empty runs full pipeline.")
    parser.add_argument('--impute', action='store_true', help="flag to create imputation values")
    args = parser.parse_args()
    etl_pipeline(stockcode=args.stockcode, impute_stockcode=args.impute)

stages: 
    etl_pipeline:
        ###
    data_drift_check:
        # dvc在此阶段将运行的主要命令
        cmd: > python src/data_handling/report_data_drift.py data/processed/processed_df.csv data/processed_df_${params.stockcode}.parquet reports/data_drift_report_${params.stockcode}.html metrics/data_drift_${params.stockcode}.json ${params.stockcode}

        # 传递给参数的默认值（在param.yaml文件中定义）
        params:
        - params.stockcode
        # 运行主命令所必需的依赖项
        deps:
        - src/data_handling/report_data_drift.py
        - src/
        # dvc跟踪的输出文件路径
        plots:
        - reports/data_drift_report_${params.stockcode}.html:
        metrics:
        - metrics/data_drift_${params.stockcode}.json:
        type: json

params:
    stockcode: <STOCKCODE OF CHOICE>

import os
import sys
import json
import pandas as pd
import datetime
from dotenv import load_dotenv
from evidently import Dataset, DataDefinition, Report
from evidently.presets import DataDriftPreset
from evidently.ui.workspace import CloudWorkspace

import src.data_handling.scripts as scripts
from src._utils import main_logger

if __name__ == '__main__':
    # 初始化evently云工作区
    load_dotenv(override=True)
    ws = CloudWorkspace(token=os.getenv('EVENTLY_API_TOKEN'), url='https://app.evidently.cloud')
    
    # 检索evently项目
    project = ws.get_project('EVENTLY AI PROJECT ID')
    
    # 从命令行参数中检索路径
    REFERENCE_DATA_PATH = sys.argv[1]
    CURRENT_DATA_PATH = sys.argv[2]
    REPORT_OUTPUT_PATH = sys.argv[3]
    METRICS_OUTPUT_PATH = sys.argv[4]
    STOCKCODE = sys.argv[5]

    # 如果不存在则创建文件夹
    os.makedirs(os.path.dirname(REPORT_OUTPUT_PATH), exist_ok=True)
    os.makedirs(os.path.dirname(METRICS_OUTPUT_PATH), exist_ok=True)

    # 提取数据集
    reference_data_full = pd.read_csv(REFERENCE_DATA_PATH)
    reference_data_stockcode = reference_data_full[reference_data_full['stockcode'] == STOCKCODE]
    current_data_stockcode = pd.read_parquet(CURRENT_DATA_PATH)

    # 定义数据模式
    nums, cats = scripts.categorize_num_cat_cols(df=reference_data_stockcode)
    for col in nums:
        current_data_stockcode[col] = pd.to_numeric(current_data_stockcode[col], errors='coerce')
    schema = DataDefinition(numerical_columns=nums, categorical_columns=cats)

    # 使用数据模式定义evently数据集
    eval_data_1 = Dataset.from_pandas(reference_data_stockcode, data_definition=schema)
    eval_data_2 = Dataset.from_pandas(current_data_stockcode, data_definition=schema)

    # 执行漂移检测
    report = Report(metrics=[DataDriftPreset()])
    data_eval = report.run(reference_data=eval_data_1, current_data=eval_data_2)
    data_eval.save_html(REPORT_OUTPUT_PATH)

    # 为dvc跟踪创建指标
    report_dict = json.loads(data_eval.json())
    num_drifts = report_dict['metrics'][0]['value']['count']
    shared_drifts = report_dict['metrics'][0]['value']['share']
    metrics = dict(
        drift_detected=bool(num_drifts > 0.0),
        num_drifts=num_drifts,
        shared_drifts=shared_drifts,
        num_cols=nums,
        cat_cols=cats,
        stockcode=STOCKCODE,
        timestamp=datetime.datetime.now().isoformat(),
    )

    # 加载指标文件
    with open(METRICS_OUTPUT_PATH, 'w') as f:
        json.dump(metrics, f, indent=4)

    main_logger.info(f'... drift metrics saved to {METRICS_OUTPUT_PATH}... ')

    # 如果发现数据漂移，则停止系统
    if num_drifts > 0.0:
        sys.exit('❌ FATAL: data drift detected. stopping pipeline')

{
    "drift_detected": false,
    "num_drifts": 0.0,
    "shared_drifts": 0.0,
    "num_cols": [
        "invoiceno",
        "invoicedate",
        "unitprice",
        "product_avg_quantity_last_month",
        "product_max_price_all_time",
        "unitprice_vs_max",
        "unitprice_to_avg",
        "unitprice_squared",
        "unitprice_log"
    ],
    "cat_cols": [
        "stockcode",
        "customerid",
        "country",
        "year",
        "year_month",
        "day_of_week",
        "is_registered"
    ],
    "timestamp": "2025-10-07T00:24:29.899495"
}

stages:
    etl_pipeline:
        ###
    data_drift_check:
        ### 
    preprocess:
        cmd: > python src/data_handling/preprocess.py --target_col ${params.target_col} --should_scale ${params.should_scale} --verbose ${params.verbose}

        deps:
        - src/data_handling/preprocess.py
        - src/data_handling/
        - src/_utils
        # params from params.yaml
        params:
        - params.target_col
        - params.should_scale
        - params.verbose
        outs:
        # train, val, test datasets
        - data/x_train_df.parquet
        - data/x_val_df.parquet
        - data/x_test_df.parquet
        - data/y_train_df.parquet
        - data/y_val_df.parquet
        - data/y_test_df.parquet
        # preprocessed input datasets
        - data/x_train_processed.parquet
        - data/x_val_processed.parquet
        - data/x_test_processed.parquet
        # trained preprocessor and human readable feature names for shap analysis
        - preprocessors/column_transformer.pkl
        - preprocessors/feature_names.json

params:
    target_col: "quantity"
    should_scale: True
    verbose: False

import os
import argparse
import json
import joblib
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
import src.data_handling.scripts as scripts
from src._utils import main_logger

def preprocess(stockcode: str = '', target_col: str = 'quantity', should_scale: bool = True, verbose: bool = False):
    # 初始化要跟踪的指标 (dvc)
    DATA_DRIFT_METRICS_PATH = os.path.join('metrics', f'data_drift_{args.stockcode}.json')

    if os.path.exists(DATA_DRIFT_METRICS_PATH):
        with open(DATA_DRIFT_METRICS_PATH, 'r') as f:
            metrics = json.load(f)
    else:
        metrics = dict()

    # 从dvc缓存加载处理后的df
    PROCESSED_DF_PATH = os.path.join('data', 'processed_df.parquet')
    df = pd.read_parquet(PROCESSED_DF_PATH)

    # 分类数值和分类列
    num_cols, cat_cols = scripts.categorize_num_cat_cols(df=df, target_col=target_col)

    if verbose:
        main_logger.info(f'num_cols: {num_cols} 
cat_cols: {cat_cols}')

    # 结构化分类列
    if cat_cols:
        for col in cat_cols:
            df[col] = df[col].astype('string')

    # 初始化预处理器（要么从dvc缓存加载，要么从头创建）
    PREPROCESSOR_PATH = os.path.join('preprocessors', 'column_transformer.pkl')
    try:
        preprocessor = joblib.load(PREPROCESSOR_PATH)
    except:
        preprocessor = scripts.create_preprocessor(num_cols=num_cols if should_scale else [], cat_cols=cat_cols)

    # 创建训练、验证、测试数据集
    y = df[target_col]
    X = df.copy().drop(target_col, axis='columns')

    # 分割
    test_size, random_state = 50000, 42
    X_tv, X_test, y_tv, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state, shuffle=False)
    X_train, X_val, y_train, y_val = train_test_split(X_tv, y_tv, test_size=test_size, random_state=random_state, shuffle=False)

    # 存储训练、验证、测试数据集 (dvc 跟踪)
    X_train.to_parquet('data/x_train_df.parquet', index=False)
    X_val.to_parquet('data/x_val_df.parquet', index=False)
    X_test.to_parquet('data/x_test_df.parquet', index=False)
    y_train.to_frame(name=target_col).to_parquet('data/y_train_df.parquet', index=False)
    y_val.to_frame(name=target_col).to_parquet('data/y_val_df.parquet', index=False)
    y_test.to_frame(name=target_col).to_parquet('data/y_test_df.parquet', index=False)

    # 预处理
    X_train = preprocessor.fit_transform(X_train)
    X_val = preprocessor.transform(X_val)
    X_test = preprocessor.transform(X_test)

    # 存储预处理后的输入数据 (dvc 跟踪)
    pd.DataFrame(X_train).to_parquet(f'data/x_train_processed.parquet', index=False)
    pd.DataFrame(X_val).to_parquet(f'data/x_val_processed.parquet', index=False)
    pd.DataFrame(X_test).to_parquet(f'data/x_test_processed.parquet', index=False)

    # 保存特征名称 (dvc 跟踪) 以用于shap分析
    with open('preprocessors/feature_names.json', 'w') as f:
        feature_names = preprocessor.get_feature_names_out()
        json.dump(feature_names.tolist(), f)

    return X_train, X_val, X_test, y_train, y_val, y_test, preprocessor

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='run data preprocessing')
    parser.add_argument('--stockcode', type=str, default='', help='specific stockcode')
    parser.add_argument('--target_col', type=str, default='quantity', help='the target column name')
    parser.add_argument('--should_scale', type=bool, default=True, help='flag to scale numerical features')
    parser.add_argument('--verbose', type=bool, default=False, help='flag for verbose logging')
    args = parser.parse_args()

    X_train, X_val, X_test, y_train, y_val, y_test, preprocessor = preprocess(
        target_col=args.target_col,
        should_scale=args.should_scale,
        verbose=args.verbose,
        stockcode=args.stockcode,
    )