企业官网建设流程全解析

• 大数据
• 批处理
• 流处理
• 数据工程

模型评估是 ML 生命周期中不可或缺的一环：它让你在未见过的数据集上基准测试模型的性能，提取选定指标、生成可视化、记录元数据，并横向对比不同模型的优劣。本文基于 Apache Beam 官方文档（website/www/site/content/en/documentation/ml/model-evaluation.md）与配套端到端示例 examples/notebooks/beam-ml/tfma_beam.ipynb，系统讲解如何通过 Beam 的ExtractEvaluateAndWriteResultsPTransform，把基于 TensorFlow Model Analysis（TFMA）的模型评估直接跑进 Beam 管道里，掌握配置评估参数、执行分布式评估、可视化指标以及一次管道对比多个模型的方法。

为什么要在 Beam 管道内做模型评估

在你的 MLOps 生态中，模型评估步骤对监控模型（或一组模型）的演进至关重要：当数据集随时间增长或发生变化、当你重新训练模型时，都需要持续评估来确认模型性能没有退化。模型评估的核心价值包括：

• 基准测试（Benchmark）：将模型性能与一个未见过的数据集进行对比，量化泛化能力；
• 指标提取与可视化：抽取你关心的指标（如均方误差、平均绝对误差），并生成可交互的图表；
• 元数据记录：将评估配置、指标、图（plots）等结果落盘保存，形成可追溯的评估记录；
• 多模型对比：在相同数据切片上比较不同模型的性能，为上线选型提供依据。

Apache Beam 提供的能力是：直接在管道内部对 TensorFlow 模型运行评估。核心是名为ExtractEvaluateAndWriteResults的 PTransform——它属于 TensorFlow Model Analysis（TFMA），这是一个用于在不同数据切片（slices of data）上执行模型评估的库。TFMA 借助 Beam 对海量数据进行分布式计算，评估结果可以在 Jupyter 或 Colab 笔记本中按数据切片对比并可视化。

在 Beam ML 的整体工作流中，这一步对应官方 About Beam ML 文档所描述的Model validation（模型验证）阶段——在部署前验证模型的性能与准确性。也就是说，模型评估与数据验证、数据预处理、模型训练、模型部署共同构成完整的 AI/ML 流水线。

认识核心 PTransform：ExtractEvaluateAndWriteResults

ExtractEvaluateAndWriteResults是 TFMA 提供的一个复合 PTransform，它把三个环节合并为一步：

1. Extract（提取）：从输入数据中提取模型推理所需的特征与标签；
2. Evaluate（评估）：按eval_config中声明的指标和切片规则执行分布式评估；
3. WriteResults（写结果）：将评估输出（配置文件、指标、plots 等）写入你指定的output_path。

由于它本身就是一个标准的 Beam PTransform，你可以像使用任何其他变换一样，把它接在数据源之后、放进beam.Pipeline上下文里运行，并自由切换 Direct Runner、Dataflow 等不同执行引擎，从而实现大规模数据上的分布式模型评估。

实操一：定义 TFMA 评估配置

评估的第一步是定义EvalConfig（评估配置），它声明三部分信息：模型信息（model_specs）、指标规格（metrics_specs），以及可选的数据切片规格（slicing_specs）。官方文档给出了一个针对线性回归模型的完整配置示例：

from google.protobuf import text_format # Define the TFMA evaluation configuration eval_config = text_format.Parse(""" ## Model information model_specs { # For keras and serving models, you need to add a `label_key`. label_key: "output" } ## This post-training metric information is merged with any built-in ## metrics from training metrics_specs { metrics { class_name: "ExampleCount" } metrics { class_name: "MeanAbsoluteError" } metrics { class_name: "MeanSquaredError" } metrics { class_name: "MeanPrediction" } } slicing_specs {} """, tfma.EvalConfig())

对配置中各字段的理解（结合 notebook 中的实践）：

• model_specs：声明待评估的模型。对 Keras 模型与 serving 模型，必须添加label_key，它指定输入记录中携带真实标签（ground truth）的字段名。上面的示例中，label_key: "output"意味着 TFRecords 中名为output的浮点特征就是标签（这一点在 notebook 写 TFRecords 时被严格执行：每条记录写入"inputs"与"output"两个 feature）。

• metrics_specs：声明评估指标。示例选择了回归任务最常用的四类指标：

  • ExampleCount：样本计数，用于了解每个切片上参与评估的样本规模；
  • MeanAbsoluteError（平均绝对误差）；
  • MeanSquaredError（均方误差）；
  • MeanPrediction（预测均值），用于观察模型输出的整体分布。

  这些 post-training 指标信息会与训练时内置（built-in）的指标合并计算。

• slicing_specs {}：声明数据切片规则。留空表示不做切片，即对全量数据计算整体指标。若需按特征切片（例如按钻石的cut、color分组评估），可在其中添加切片特征，TFMA 会为每个切片单独产出指标，便于发现模型在特定子群体上的短板。

注意：eval_config通过text_format.Parse从 Protocol Buffers 文本格式解析为tfma.EvalConfig()对象，这是 TFMA 配置的标准写法，请确保google.protobuf可用。

实操二：构建评估管道

定义好配置后，创建管道运行评估。官方文档给出的管道骨架如下：

from tfx_bsl.public import tfxio eval_shared_model = tfma.default_eval_shared_model( eval_saved_model_path='model_path', eval_config=eval_config) tfx_io = tfxio.TFExampleRecord( file_pattern='tfrecords_path', raw_record_column_name=tfma.ARROW_INPUT_COLUMN) # Run evaluation with beam.Pipeline() as pipeline: _ = ( pipeline | 'ReadData' >> tfx_io.BeamSource() | 'EvalModel' >> tfma.ExtractEvaluateAndWriteResults( eval_shared_model=eval_shared_model, eval_config=eval_config, output_path='output_path'))

逐段拆解这段代码：

• tfma.default_eval_shared_model：根据已保存的模型路径（eval_saved_model_path）与eval_config构造评估共享模型对象。它负责在分布式 worker 上加载模型并执行推理；
• tfxio.TFExampleRecord：来自tfx_bsl.public.tfxio的 TFXIO 数据读取器。它声明输入数据为 TFRecord 格式（file_pattern='tfrecords_path'），并通过raw_record_column_name=tfma.ARROW_INPUT_COLUMN指定把原始tf.Example记录作为 Arrow 输入列保留下来，供 TFMA 的模型输入与标签提取使用；
• tfx_io.BeamSource()：将 TFXIO 读取器封装为 Beam 的 PCollection 数据源，作为管道的输入；
• tfma.ExtractEvaluateAndWriteResults(...)：传入三个关键参数：
  • eval_shared_model：评估用的共享模型；
  • eval_config：评估配置；
  • output_path：结果输出目录。管道运行结束后，该路径下会保存包括**配置文件（config file）、指标（metrics）、图（plots）**在内的全部评估结果。

实操三：加载结果并可视化

评估管道把结果写入output_path后，即可在笔记本中加载并渲染可视化：

# Visualize the results. result = tfma.load_eval_result(output_path=output_path) tfma.view.render_slicing_metrics(result)

• tfma.load_eval_result(output_path=...)从输出路径加载评估结果；
• tfma.view.render_slicing_metrics(result)渲染按切片分布的指标可视化，适合在 Jupyter / Colab 中交互式查看不同数据切片上的指标表现。

端到端示例：钻石价格预测模型的评估

仓库自带的 tfma_beam.ipynb 提供了一个完整、可复现的端到端示例，覆盖"数据准备 → 训练模型 → Beam 评估 → 可视化 → 多模型对比"全流程。下面按照 notebook 的实际步骤展开。

环境准备

本地 Jupyter 运行前，需先启用 TFMA 所需的 Jupyter 扩展：

jupyter nbextension enable --py widgetsnbextension --sys-prefix jupyter nbextension install --py --symlink tensorflow_model_analysis --sys-prefix jupyter nbextension enable --py tensorflow_model_analysis --sys-prefix

然后安装依赖（安装 TFMA 会自动拉取其全部依赖）：

pip install -U pip pip install tensorflow-model-analysis

安装后需重启运行时。notebook 中注明该配置在 Colab 下使用 TensorFlow 2.11、TFMA 0.43、Apache Beam 2.44 验证通过，并兼容当前发布版本；同时要求 Python 3。建议安装后先打印tf.__version__、beam.__version__、tfma.__version__确认各组件版本无误再继续。

数据准备与 TFRecords 生成

示例使用 TFDS（TensorFlow Datasets）自带的diamonds 钻石数据集（包含 53,940 颗钻石的物理属性，如重量 carat、切工 cut、颜色 color、净度 clarity，以及价格 price），训练一个预测钻石价格的线性回归模型，并用均方误差、平均绝对误差等指标评估。

数据侧的关键步骤：

1. 划分数据：为模拟"模型性能随数据增长而提升"的场景，用数据集前 40% 训练 v1 模型，再用前 80% 训练 v2 模型，另切出独立的测试集（train[80%:90%]）与验证集（train[90%:]）：

   (ds_train_v1, ds_test, ds_val), info = tfds.load( 'diamonds', split=['train[:40%]', 'train[80%:90%]', 'train[90%:]'], as_supervised=True, with_info=True)

2. 特征工程：数值特征（carat、depth、table、x、y、z）共 6 维，类别特征（color、cut、clarity）用tf.one_hot编码，最终拼成完整特征向量；模型输出为单个标量（价格），即output。
3. 写出 TFRecords：TFMA 与 Beam 需要从文件读取评估数据，因此把验证集序列化为tf.train.Example写入data/val_data.tfrecord，其中每个 example 包含"inputs"（特征 float_list）与"output"（标签 float_list）两个 feature——这正是eval_config里label_key: "output"能正确取到标签的前提。

训练与保存模型

示例通过construct_model(model_name, train_data)构建一个含归一化层与单隐层（8 个神经元，ReLU 激活）的 Keras 模型，用 Adam（学习率 0.1）+mean_absolute_error损失训练 5 个 epoch，然后把模型保存到磁盘：

model_path_v1 = 'saved_model_v1' model_v1.save(model_path_v1)

保存后的 SavedModel 路径将作为eval_saved_model_path传给 TFMA。

在 Beam 管道中执行评估

评估部分与"实操一/二"完全一致：先定义eval_config（采用ExampleCount、MeanAbsoluteError、MeanSquaredError、MeanPrediction四个指标），再构建管道：

from tfx_bsl.public import tfxio output_path = 'evaluation_results' eval_shared_model = tfma.default_eval_shared_model( eval_saved_model_path=model_path_v1, eval_config=eval_config) tfx_io = tfxio.TFExampleRecord( file_pattern=tfrecord_file, raw_record_column_name=tfma.ARROW_INPUT_COLUMN) # Run Evaluation. with beam.Pipeline() as pipeline: _ = ( pipeline | 'ReadData' >> tfx_io.BeamSource() | 'EvalModel' >> tfma.ExtractEvaluateAndWriteResults( eval_shared_model=eval_shared_model, eval_config=eval_config, output_path=output_path))

随后加载结果并渲染切片指标：

result = tfma.load_eval_result(output_path=output_path) tfma.view.render_slicing_metrics(result)

进阶：在单次 Beam 管道中对比多个模型

生产环境选型时往往需要横向比较多个候选模型。Beam + TFMA 的另一个亮点是可以在一次管道运行中同时评估并对比多个模型。notebook 展示了训练第二个模型（v2，使用前 80% 数据训练，模拟数据增多后模型性能提升）后的对比流程。

对比版评估配置

对比场景的eval_config_compare与单模型版本的关键差异在于model_specs部分——声明两个模型，并为基准模型设置is_baseline: true：

eval_config_compare = text_format.Parse(""" ## Model information model_specs { name: "model_v1" # For keras (and serving models), add a `label_key`. label_key: "output" is_baseline: true } model_specs { name: "model_v2" # For keras (and serving models), add a `label_key`. label_key: "output" } ## This post-training metric information is merged with any built-in ## metrics from training. metrics_specs { metrics { class_name: "ExampleCount" } metrics { class_name: "MeanAbsoluteError" } metrics { class_name: "MeanSquaredError" } metrics { class_name: "MeanPrediction" } } slicing_specs {} """, tfma.EvalConfig())

要点：

• 每个model_specs通过name标识模型；is_baseline: true指定 v1 为基准模型，TFMA 会以它为参照展示其他模型的相对差异；
• 两个模型仍使用同一份验证数据（同一个tfrecord_file），保证对比口径一致。

多模型评估管道

相应地，eval_shared_model参数变为模型列表，每个模型通过model_name与配置中的名字对应：

eval_shared_models = [ tfma.default_eval_shared_model( model_name='model_v1', eval_saved_model_path=model_path_v1, eval_config=eval_config_compare), tfma.default_eval_shared_model( model_name='model_v2', eval_saved_model_path=model_path_v2, eval_config=eval_config_compare), ] tfx_io = tfxio.TFExampleRecord( file_pattern=tfrecord_file, raw_record_column_name=tfma.ARROW_INPUT_COLUMN) # Run the evaluation. with beam.Pipeline() as pipeline: _ = ( pipeline | 'ReadData' >> tfx_io.BeamSource() | 'EvalModel' >> tfma.ExtractEvaluateAndWriteResults( eval_shared_model=eval_shared_models, eval_config=eval_config_compare, output_path=output_path_compare))

对比结果可视化

加载对比结果使用load_eval_results（复数形式，传入输出路径列表），并调用render_time_series渲染时间序列视图——默认展示验证集规模随时间的变化，你还可以通过add metric series勾选 loss 与 mean absolute error 等指标，直观观察模型性能随数据量增长的趋势：

results = tfma.load_eval_results(output_paths=output_path_compare) tfma.view.render_time_series(results)

这种"同一管道、同一份数据、多模型并行评估"的能力，使 Beam + TFMA 特别适合模型迭代监控、A/B 候选评估与生产前选型验证。

小结与延伸阅读

总结一下，在 Apache Beam 中完成 TensorFlow 模型评估的标准路径是：

1. 定义EvalConfig：声明model_specs（含label_key）、metrics_specs（指标类名）与slicing_specs（切片规则）；
2. 构造评估输入：用tfma.default_eval_shared_model包装已保存的模型，用tfxio.TFExampleRecord+BeamSource()读取 TFRecords 评估数据；
3. 运行管道：将ExtractEvaluateAndWriteResults作为 PTransform 接入beam.Pipeline，指定output_path落盘结果；
4. 可视化与对比：load_eval_result+render_slicing_metrics查看单模型切片指标；多模型场景使用is_baseline、eval_shared_models列表以及load_eval_results+render_time_series。

由于ExtractEvaluateAndWriteResults是标准 Beam PTransform，整套评估逻辑可以无缝迁移到不同 Runner（如 Dataflow）上，在海量数据上获得分布式评估能力。

想深入实践，可以直接运行仓库中的 tfma_beam.ipynb（包含从 TFDS 创建 tfrecords、训练模型到 Beam 评估的完整可执行代码）；了解 Beam ML 的整体定位与模型验证在整个 AI/ML 工作流中的位置，可阅读 About Beam ML 的 Model validation 一节；关于 TFMA 支持的更多指标与可视化参数，可查阅 TFMA 官方指标与 plots 文档。

• 大数据
• 批处理
• 流处理
• 数据工程