构建基于DVC服务网格与Angular的机器学习特征全链路可观测性与版本化实践

MLOps

文章字数: 3k

阅读时长: 12 分

团队的机器学习模型迭代开始失控，根源不在算法，而在特征。一个特征工程脚本的微小改动，一周后可能导致线上模型性能的神秘衰减，而彼时，没人能准确复盘出是哪个版本的代码、处理了哪一批数据、最终生成了线上正在使用的那版特征。特征的生成过程变成了一个难以追溯、无法复现的黑盒，这在生产环境中是致命的。我们的目标是构建一个平台，不仅要能版本化管理特征工程的“代码”，更要能版本化管理其“数据”和“制品”，并为整个流程提供健壮的部署与观测能力。

初步构想是建立一个端到端的特征管理平台，它需要一个清晰的Web界面（控制平面）、一套可靠的特征版本化机制，以及一个支持渐进式发布和精细化监控的运行时环境。

技术选型决策过程充满了权衡：

特征版本化 (DVC): 我们没有选择简单的Git LFS。Git LFS能处理大文件，但它无法理解数据处理的“依赖关系图”。DVC（Data Version Control）的核心优势在于它能通过dvc.yaml文件定义数据处理的阶段（stages）和依赖（dependencies），形成一个可重复执行的DAG（有向无环图）。这意味着我们可以精确地将feature_script.py的v1.2版本和它处理过的raw_data/clicks_20231026.csv以及产出的features/user_profile_v3.parquet绑定在一起。这种对“代码-数据-制品”三位一体的版本控制能力，是实现可复现性的基石。
运行时与部署 (Service Mesh - Istio): 特征计算服务是典型的微服务。新版本的特征逻辑上线风险极高，全量发布是不可接受的。我们需要精细的流量控制能力，例如“将organization_id为‘test-corp’的内部请求路由到新版特征服务，其他流量走老版本”。这种基于请求内容的路由是服务网格的强项。我们选择Istio，因为它提供了开箱即用的金丝雀发布、流量镜像、mTLS加密以及丰富的遥测数据（Metrics, Traces），而这一切对应用代码是透明的。
控制平面 (Angular): 团队前端技术栈以Angular为主，其强类型的TypeScript、依赖注入系统和模块化的架构非常适合构建复杂的、需要长期维护的企业级单页应用。这个控制平面需要展示复杂的特征依赖图、部署状态、版本历史和来自Istio的性能指标，Angular的健壮性是完成这个任务的保障。
前端测试 (Vitest): 传统上Angular项目使用Karma和Jasmine，但它们的启动速度和配置复杂性一直是个痛点。Vitest基于Vite，提供了几乎瞬时的热更新和测试执行速度，其与Jest兼容的API也让迁移成本很低。在一个追求高迭代效率的平台项目中，开发体验至关重要，Vitest能显著提升前端测试编写和调试的效率。

步骤化实现：从版本化到可观测部署

1. DVC定义可复现的特征工程流水线

我们的核心是dvc.yaml，它定义了整个特征生成过程。假设我们有一个从原始点击流日志中提取用户画像特征的任务。

项目结构如下：

.
├── dvc.yaml
├── params.yaml
├── data/
│   └── raw_clicks.csv
├── src/
│   ├── preprocess.py
│   └── feature_engineering.py
└── dvc.lock

params.yaml 存放超参数，使其与代码分离：

# params.yaml
preprocess:
  min_session_duration: 5
  columns_to_drop: ['user_agent', 'ip_address']

feature_engineering:
  time_window_hours: 72
  top_n_categories: 10

dvc.yaml 定义了两个阶段：预处理和特征工程。注意deps（依赖）、params（参数）和outs（输出）的声明，DVC会根据这些元数据计算哈希值来判断是否需要重新执行。

# dvc.yaml
stages:
  preprocess:
    cmd: python src/preprocess.py data/raw_clicks.csv data/processed_clicks.parquet
    deps:
      - data/raw_clicks.csv
      - src/preprocess.py
    params:
      - preprocess.min_session_duration
      - preprocess.columns_to_drop
    outs:
      - data/processed_clicks.parquet

  build_features:
    cmd: python src/feature_engineering.py data/processed_clicks.parquet data/features/user_profile_v1.parquet
    deps:
      - data/processed_clicks.parquet
      - src/feature_engineering.py
    params:
      - feature_engineering.time_window_hours
      - feature_engineering.top_n_categories
    outs:
      - data/features/user_profile_v1.parquet

当数据科学家修改了src/feature_engineering.py中的逻辑，只需运行dvc repro，DVC会自动检测到build_features阶段的依赖已改变，并重新执行该阶段，生成新的特征文件。所有这些变更（代码、参数、数据哈希）都被提交到Git中，而数据文件本身则通过dvc push推送到S3等远端存储。这就是可复现性的核心。

2. 服务网格实现特征服务的金丝雀发布

特征生成后，需要通过一个实时服务API对外提供。我们有两个版本的特征服务部署在Kubernetes中：feature-server:v1 和 feature-server:v2。

首先，定义一个通用的Service和DestinationRule来管理所有版本的服务子集。

# feature-server-destinationrule.yaml
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: feature-server-dr
spec:
  host: feature-server
  subsets:
    - name: v1
      labels:
        version: v1
    - name: v2
      labels:
        version: v2
---
apiVersion: v1
kind: Service
metadata:
  name: feature-server
  labels:
    app: feature-server
spec:
  ports:
  - port: 8080
    name: http
  selector:
    app: feature-server

关键在于VirtualService的配置。下面的配置实现了两个目标：

默认情况下，90%的流量流向v1版本，10%的流量流向v2版本，实现小范围灰度。
如果HTTP请求头中包含x-user-group: internal-testers，则100%将该请求路由到v2版本。这允许我们在不影响外部用户的情况下对新版本进行内部测试。

# feature-server-virtualservice.yaml
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: feature-server-vs
spec:
  hosts:
    - "feature-server.mlops.svc.cluster.local"
    - "features.internal.mycorp.com"
  gateways:
    - mesh
    - my-internal-gateway
  http:
    - match:
        - headers:
            x-user-group:
              exact: internal-testers
      route:
        - destination:
            host: feature-server
            subset: v2
          weight: 100
    - route:
        - destination:
            host: feature-server
            subset: v1
          weight: 90
        - destination:
            host: feature-server
            subset: v2
          weight: 10
      # 为v2流量添加故障注入，模拟5%的请求出现503错误，测试客户端容错能力
      fault:
        abort:
          httpStatus: 503
          percentage:
            value: 5.0
      # 增加超时和重试策略，增强服务韧性
      timeout: 5s
      retries:
        attempts: 3
        perTryTimeout: 2s

这里的fault和retries配置展示了服务网格的另一大价值：无需修改任何应用代码，即可在基础设施层实现复杂的韧性工程策略。

3. Angular控制平面与状态管理

Angular应用的核心是服务和组件。我们创建一个FeaturePipelineService来与后端API交互，获取DVC流水线和Istio部署的状态。

// src/app/core/services/feature-pipeline.service.ts
import { Injectable } from '@angular/core';
import { HttpClient, HttpErrorResponse } from '@angular/common/http';
import { Observable, throwError, timer } from 'rxjs';
import { catchError, retry, switchMap, shareReplay } from 'rxjs/operators';

export interface PipelineStage {
  name: string;
  command: string;
  dependencies: string[];
  outputs: string[];
  status: 'running' | 'succeeded' | 'failed' | 'unchanged';
}

export interface DeploymentStatus {
  version: string;
  trafficWeight: number;
  instanceCount: number;
  metrics: { rps: number; errorRate: number; latencyP99: number };
}

@Injectable({
  providedIn: 'root',
})
export class FeaturePipelineService {
  private readonly API_BASE = '/api/v1/pipelines';
  
  // 使用shareReplay来缓存并共享HTTP请求结果，避免重复请求
  public deploymentStatus$: Observable<DeploymentStatus[]>;

  constructor(private http: HttpClient) {
    // 轮询获取部署状态，生产环境中可能会使用WebSocket替代
    this.deploymentStatus$ = timer(0, 5000).pipe(
      switchMap(() => this.http.get<DeploymentStatus[]>(`${this.API_BASE}/my-feature/deployments`)),
      retry({ count: 3, delay: 2000 }), // 简单的重试策略
      catchError(this.handleError),
      shareReplay(1)
    );
  }

  getPipelineGraph(pipelineId: string): Observable<PipelineStage[]> {
    return this.http.get<PipelineStage[]>(`${this.API_BASE}/${pipelineId}/graph`).pipe(
      catchError(this.handleError)
    );
  }

  // 一个健壮的错误处理函数
  private handleError(error: HttpErrorResponse) {
    console.error(
      `Backend returned code ${error.status}, body was: `, error.error
    );
    // 向上传递一个用户友好的错误信息
    return throwError(() => new Error('Failed to fetch pipeline data; please try again later.'));
  }
}

组件层面，我们使用Mermaid.js来可视化DVC生成的依赖图。

// src/app/features/pipeline-viewer/pipeline-viewer.component.ts
import { Component, OnInit, ElementRef, ViewChild, AfterViewInit, Input } from '@angular/core';
import mermaid from 'mermaid';
import { FeaturePipelineService, PipelineStage } from '../../core/services/feature-pipeline.service';

@Component({
  selector: 'app-pipeline-viewer',
  template: `<div #mermaidContainer class="mermaid-graph"></div>`,
  styles: [`.mermaid-graph { text-align: center; }`],
})
export class PipelineViewerComponent implements OnInit, AfterViewInit {
  @Input() pipelineId!: string;
  @ViewChild('mermaidContainer') private mermaidContainer!: ElementRef;

  constructor(private pipelineService: FeaturePipelineService) {}

  ngOnInit(): void {
    if (!this.pipelineId) {
      console.error("Pipeline ID is required for viewer component.");
      return;
    }
    this.pipelineService.getPipelineGraph(this.pipelineId).subscribe(stages => {
      const graphDefinition = this.buildMermaidGraph(stages);
      this.renderGraph(graphDefinition);
    });
  }
  
  ngAfterViewInit(): void {
    mermaid.initialize({ startOnLoad: false, theme: 'neutral' });
  }

  private buildMermaidGraph(stages: PipelineStage[]): string {
    let graph = 'graph TD;\n';
    stages.forEach(stage => {
      stage.dependencies.forEach(dep => {
        // 美化节点名称，去除路径
        const depNode = dep.split('/').pop()?.replace('.', '_');
        const stageNode = stage.name;
        graph += `    ${depNode}[${dep}] --> ${stageNode}(${stageNode});\n`;
      });
      stage.outputs.forEach(out => {
        const outNode = out.split('/').pop()?.replace('.', '_');
        const stageNode = stage.name;
        graph += `    ${stageNode} --> ${outNode}[${out}];\n`;
      });
    });
    return graph;
  }
  
  private renderGraph(graphDefinition: string) {
    if (this.mermaidContainer) {
      mermaid.render('graphDiv', graphDefinition, (svgCode) => {
        this.mermaidContainer.nativeElement.innerHTML = svgCode;
      });
    }
  }
}

4. Vitest保障前端组件的可靠性

对FeaturePipelineService进行单元测试，确保其HTTP逻辑、错误处理和重试机制按预期工作。这里使用msw (Mock Service Worker) 或类似的库来拦截和模拟HTTP请求会更健壮，但为简化示例，我们直接模拟HttpClient。

// src/app/core/services/feature-pipeline.service.spec.ts
import { TestBed } from '@angular/core/testing';
import { HttpClientTestingModule, HttpTestingController } from '@angular/common/http/testing';
import { describe, it, expect, afterEach, vi } from 'vitest';
import { FeaturePipelineService, DeploymentStatus } from './feature-pipeline.service';
import { fakeAsync, tick } from '@angular/core/testing';

describe('FeaturePipelineService', () => {
  let service: FeaturePipelineService;
  let httpMock: HttpTestingController;
  const API_BASE = '/api/v1/pipelines';

  beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [HttpClientTestingModule],
      providers: [FeaturePipelineService],
    });
    service = TestBed.inject(FeaturePipelineService);
    httpMock = TestBed.inject(HttpTestingController);
  });

  afterEach(() => {
    httpMock.verify(); // 确保没有未处理的请求
  });

  it('should be created', () => {
    expect(service).toBeTruthy();
  });

  it('deploymentStatus$ should poll and fetch deployment status', fakeAsync(() => {
    const mockStatus: DeploymentStatus[] = [
      { version: 'v1', trafficWeight: 100, instanceCount: 3, metrics: { rps: 150, errorRate: 0.01, latencyP99: 120 } },
    ];
    let result: DeploymentStatus[] | undefined;

    service.deploymentStatus$.subscribe(data => {
      result = data;
    });

    // 第一次轮询 (timer(0, ...))
    const req = httpMock.expectOne(`${API_BASE}/my-feature/deployments`);
    expect(req.request.method).toBe('GET');
    req.flush(mockStatus);
    
    expect(result).toEqual(mockStatus);

    // 第二次轮询
    tick(5000);
    const req2 = httpMock.expectOne(`${API_BASE}/my-feature/deployments`);
    req2.flush([]);
    expect(result).toEqual([]);

    tick(5000); // 清理定时器
  }));

  it('should handle HTTP errors gracefully on getPipelineGraph', () => {
    const pipelineId = 'test-pipeline';
    const emsg = 'deliberate 404 error';
    const errorSpy = vi.spyOn(console, 'error');

    service.getPipelineGraph(pipelineId).subscribe({
      next: () => fail('should have failed with the 404 error'),
      error: (error) => {
        expect(error.message).toContain('Failed to fetch pipeline data');
      },
    });

    const req = httpMock.expectOne(`${API_BASE}/${pipelineId}/graph`);
    req.flush(emsg, { status: 404, statusText: 'Not Found' });
    
    expect(errorSpy).toHaveBeenCalled();
  });
});

这个测试验证了服务的轮询逻辑和核心的错误处理分支，确保UI在后端服务不稳定时不会崩溃，并能给出适当的用户提示。这对于一个面向内部开发者的平台来说，是建立信任的基础。

当前方案的局限性与未来展望

我们构建的这套体系解决了特征工程中最棘手的可复现性和安全部署问题。DVC提供了数据血缘的“静态”保障，服务网格提供了运行时流量调度的“动态”控制，而Angular和Vitest则构建了一个高效、可靠的“交互”界面。

然而，这个方案并非银弹。首先，DVC的集成对CI/CD流程有一定的侵入性，需要精心设计流水线以缓存DVC的cache，否则每次构建都会非常耗时。其次，Istio本身是一个复杂的组件，运维一个生产级的Istio集群需要专门的知识储备，对于小团队而言，其管理开销可能大于收益。可以考虑使用Linkerd等更轻量的服务网格作为替代。

未来的迭代方向很明确：一是实现基于GitOps的自动化部署，当数据科学家在特定分支上合并一个dvc.lock文件的更新时，ArgoCD等工具能自动触发对应的Kubernetes资源（Deployment, VirtualService）的更新，实现完全声明式的发布流程。二是在控制平面中集成更丰富的可观测性数据，不仅仅是Istio的L7层遥测，还应包括特征服务本身的业务指标，例如特征值的分布漂移、空值率等，形成一个从基础设施到业务数据的完整监控闭环。

MLOps DVC Service Mesh Angular Vitest Istio

使用 OpenTelemetry 与 TypeScript 构建 Storybook 组件生命周期追踪插件

2023-10-27 前端工程化

TypeScript OpenTelemetry Storybook

构建基于Knative的事件驱动型Snowflake数据摄取核心库与自动化工具链

2023-10-27 云原生

核心库 Snowflake Knative 构建与工具 Golang Serverless