【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载本文围绕 Apache Beam 官方示例文档side-input-updates.md展开讲解如何借助WatchFilePattern作为侧输入side input让RunInference在流水线持续运行期间自动刷新 ML 模型无需停机。读者将掌握WatchFilePattern的原理、ModelMetadata与PredictionResult的结构、图像预处理与后处理的完整代码以及将其迁移到 PyTorch 等框架的通用模式。背景为什么要用侧输入更新模型在典型的推理流水线中模型通过ModelHandler配置对象加载一旦流水线启动模型就固定下来。若要换模型传统做法是停止流水线、替换模型文件、再重新启动这在生产环境意味着服务中断。Apache Beam 的RunInference提供了动态模型更新机制通过一个发出ModelMetadata的侧输入PCollection可以在流水线运行期间实时更新ModelHandler中的模型。实现方式有两种使用 Beam 内置的WatchFilePattern模式自动监控目录中匹配file_pattern的文件并按时间戳挑选最新版本自定义一个侧输入PCollection自行定义模型更新的触发逻辑。本文以第一种方式为主完整呈现一个基于 TensorFlow 的图像分割/分类推理流水线。架构总览数据流与模型更新链路整个流水线由三部分组成数据源Pub/Sub Topic 持续发出UTF-8编码的图片路径字符串模型文件源WatchFilePattern监控 GCS 上匹配gs://your-bucket/*.h5的模型文件推理与更新RunInference以TFModelHandlerTensor加载模型并把WatchFilePattern的输出作为model_metadata_pcoll侧输入在流水线不停止的情况下完成模型热替换。从源码结构看WatchFilePattern是定义在 sdks/python/apache_beam/ml/inference/utils.py 中的beam.PTransformutils.py#L114其内部调用链为MatchContinuously持续匹配文件interval 间隔轮询 → AttachKey把 FileMetadata 转成 (path, metadata) 键值对 → _GetLatestFileByTimeStamp比较文件时间戳与流水线启动时间输出最新文件的 ModelMetadata → _ConvertIterToSingleton把 Iterable 折叠成单一元素保证可被 AsSingleton 消费 → WindowIntoGlobalWindows Repeatedly(AfterProcessingTime(1)) DISCARDING其中_GetLatestFileByTimeStamp通过状态CombiningValueStateSpec记录当前最大时间戳只有比流水线启动时刻更新的文件才会被输出为新的ModelMetadata见 utils.py#L86-L111_ConvertIterToSingleton则借助计数状态保证每个新文件只被当作侧输入发出一次见 utils.py#L64-L83。这两个内部 DoFn 的行为在 utils_test.py 中有对应的单元测试例如test_latest_file_with_timestamp_after_pipeline_construction_time验证了只有启动后更新的文件才会被选中。RunInference的构造参数model_metadata_pcoll定义在 base.py#L1003在expand中会通过beam.pvalue.AsSingleton包装后传入_RunInferenceDoFnbase.py#L1138-L1140。_RunInferenceDoFn.process会对比侧输入中的model_id与当前已加载模型的路径一旦发现新值就调用update_model重新加载模型并重新采集指标base.py#L1408-L1440。设置数据源Pub/Sub 发出图片路径流水线从 Pub/Sub Topic 读取数据源Topic 发出的每条消息是UTF-8编码的图片路径字符串后续用它读取并预处理图片送入推理。pubsub_topic topic_emitting_image_names消息内容示例gs://your-bucket/images/cat_001.jpg之类的路径。实际部署时把topic_emitting_image_names替换为真实的 Pub/Sub Topic 名称如projects/project/topics/topic。模型选择HDF5 格式的 TensorFlow 模型本例使用以 HDF5 格式保存的 TensorFlow 模型.h5文件。HDF5 是 Keras/TensorFlow 保存整模型结构 权重的常用格式便于在流水线运行期间以新文件整体替换。图片预处理read_image 函数Pub/Sub 发来的是图片路径需先读取并预处理成模型可接受的输入。文档给出的read_image函数如下import io from PIL import Image from apache_beam.io.filesystems import FileSystems import numpy import tensorflow as tf def read_image(image_file_name): with FileSystems().open(image_file_name, r) as file: data Image.open(io.BytesIO(file.read())).convert(RGB) img data.resize((224, 224)) img numpy.array(img) / 255.0 img_tensor tf.cast(tf.convert_to_tensor(img[...]), dtypetf.float32) return img_tensor要点说明使用FileSystems().open(...)而非本地open因此既支持 GCSgs://也支持本地路径这正是配合分布式 runner 的关键convert(RGB)统一通道数为 3缩放到 224×224 是预训练图像模型如 ImageNet 系模型的常见输入尺寸除以 255.0 归一化到 [0, 1]最终转换为tf.float32张量匹配TFModelHandlerTensor的输入类型要求。流水线代码核心步骤与侧输入接线流水线步骤从 Pub/Sub Topic 获取图片路径用read_image读取并预处理图片将图片送入RunInference其中传入model_handler与model_metadata_pcoll两个关键参数。初始化模型处理器TFModelHandlerTensormodel_handler使用TFModelHandlerTensor它位于 sdks/python/apache_beam/ml/inference/tensorflow_inference.pytensorflow_inference.py#L228专门用于处理tf.Tensor输入内部会调用tf.stack批量推理。from apache_beam.ml.inference.tensorflow_inference import TFModelHandlerTensor # initialize TFModelHandlerTensor with a .h5 model saved in a directory accessible by the pipeline. tf_model_handler TFModelHandlerTensor(model_urigs://your-bucket/model_path.h5)TFModelHandlerTensor构造参数中常用的有model_uri模型路径、model_type默认SAVED_MODEL、create_model_fn配合SAVED_WEIGHTS使用、load_model_args传给 TensorFlow 加载函数的参数、min_batch_size/max_batch_size/max_batch_duration_secs控制BatchElements的批大小与缓冲时长以及large_model大模型跨进程共享开关。从源码看update_model_path的实现只是把model_uri替换为新路径tensorflow_inference.py#L306-L307后续重新load_model即完成热更新。构建侧输入WatchFilePatternmodel_metadata_pcoll是RunInference的侧输入PCollection用于在不停止流水线的情况下更新model_handler中的模型。本例用它监控匹配.h5文件的 glob 模式file_pattern gs://your-bucket/*.h5 side_input_pcoll ( pipeline | FilePatternUpdates WatchFilePattern(file_patternfile_pattern))WatchFilePattern的构造参数见 utils.py#L114-L149参数含义默认值file_pattern本地路径或 GCSgs://路径支持*、?、[...]等 glob 字符必填interval轮询检查匹配文件的间隔秒360stop_timestamp超过该时间戳后不再检查文件MAX_TIMESTAMP使用WatchFilePattern时必须注意三个约束文件名不可复用若某个文件名已被使用过之后再次添加/更新同名文件会被忽略。要触发模型更新每次必须上传唯一文件名的新文件启动前需至少存在一个匹配文件流水线启动时刻之前file_pattern必须能匹配到至少一个文件否则流水线会失败。这个初始文件通常就是initial_model_path仅适用于流式模式该 transform 内部由MatchContinuously产生无界数据源仅支持流式streaming运行在批处理模式下运行会得到异常结果或导致流水线卡住。组装主数据流与推理model_metadata_pcoll期望的是一组兼容AsSingleton视图的ModelMetadata元素。由于这里使用WatchFilePattern它会自动完成窗口化global windowing并把输出包装成ModelMetadata无需手工处理。import apache_beam as beam from apache_beam.ml.inference.utils import WatchFilePattern from apache_beam.ml.inference.base import RunInference with beam.Pipeline() as pipeline: file_pattern gs://your-bucket/*.h5 pubsub_topic topic_emitting_image_names side_input_pcoll ( pipeline | FilePatternUpdates WatchFilePattern(file_patternfile_pattern)) images_pcoll ( pipeline | ReadFromPubSub beam.io.ReadFromPubSub(topicpubsub_topic) | DecodeBytes beam.Map(lambda x: x.decode(utf-8)) | PreProcessImage beam.Map(read_image) ) inference_pcoll ( images_pcoll | RunInference RunInference( model_handlertf_model_handler, model_metadata_pcollside_input_pcoll))触发一次模型更新的流程流水线启动并开始处理 Pub/Sub 中的图片消息观察到RunInference有输出后向 GCS bucket 上传一个匹配file_patterngs://your-bucket/*.h5的.h5模型注意文件名必须是此前未使用过的WatchFilePattern在下一个interval轮询周期发现新文件经_GetLatestFileByTimeStamp与_ConvertIterToSingleton处理后输出新的ModelMetadataRunInference的_RunInferenceDoFn.process检测到model_id变化调用_load_model→update_model_path更新TFModelHandlerTensor的model_uri并重新加载模型base.py#L1320-L1361后续批次使用新模型推理。注意侧输入更新的频率是不确定的两次更新之间可能间隔较长时间。这是因为侧输入的处理取决于 runner 对侧输入视图的刷新策略不应假设上传文件后会立即生效。后处理解析 PredictionResult 对象推理完成后RunInference输出PredictionResult对象它包含三个字段定义见 base.py#L82-L97字段含义example输入样例本例为预处理后的图片张量inference模型推理结果model_id执行本次推理所用的模型标识用于确认使用的是哪个版本的模型文档给出的PostProcessor示例通过model_id关联预测结果与模型版本并用 ImageNet 标签反查类别名from apache_beam.ml.inference.base import PredictionResult class PostProcessor(beam.DoFn): Process the PredictionResult to get the predicted label and model id used for inference. def process(self, element: PredictionResult) - typing.Iterable[str]: predicted_class numpy.argmax(element.inference[0], axis-1) labels_path tf.keras.utils.get_file( ImageNetLabels.txt, https://storage.googleapis.com/download.tensorflow.org/data/ImageNetLabels.txt ) imagenet_labels numpy.array(open(labels_path).read().splitlines()) predicted_class_name imagenet_labels[predicted_class] return predicted_class_name.title(), element.model_id post_processor_pcoll (inference_pcoll | PostProcessor PostProcessor())使用PostProcessor时注意tf.keras.utils.get_file会在运行节点下载标签文件若 runner 环境无法访问外网应改为预先将标签文件放到流水线可读的 GCS/本地路径。numpy.argmax(element.inference[0], axis-1)假定inference的 batch 维度在首位如果模型输出结构不同需按实际形状调整索引。运行流水线result pipeline.run().wait_until_finish()由于WatchFilePattern是无界数据源流水线必须在流式模式下运行。以 DataflowRunner 为例可参考仓库中的 PyTorch 示例 pytorch_image_classification_with_side_inputs.py 的命令行参数组织方式包括--streaming、--interval10、--topicpubsub_topic、--file_patternglob_pattern以及--requirements_fileapache_beam/ml/inference/tensorflow_tests_requirements.txtTensorFlow 依赖清单见 tensorflow_tests_requirements.txt。注意ModelMetadata中的model_name会作为前缀附加到RunInference计算出的指标metrics上。从源码看_RunInferenceDoFn.process在检测到模型更新时会以model_name作为前缀重建_MetricsCollectorbase.py#L1431-L1434因此每个模型版本的推理计数、批大小、延迟等指标都独立可查便于在运维中对比不同版本模型的效果。通用模式与 PyTorch 迁移本文的写法可以直接套用到其他深度学习框架。仓库中提供了一个 PyTorch 版本的可运行示例pytorch_image_classification_with_side_inputs.py其结构与本文示例一一对应数据源同样来自 Pub/Sub发出图片路径模型处理器换成PytorchModelHandlerTensor初始模型默认gs://apache-beam-samples/run_inference/resnet152.pthWatchFilePattern(intervalknown_args.interval, file_patternknown_args.file_pattern)作为model_metadata_pcoll传入RunInference后处理PostProcessor输出文件名、预测类别与model_id三元组。该示例还展示了KeyedModelHandler的用法当输入是键值对key 为文件名时用KeyedModelHandler包裹底层ModelHandler即可在带键输入的流水线中同样使用侧输入完成模型热更新。总结与最佳实践不停机更新用WatchFilePattern作为model_metadata_pcoll侧输入RunInference会在流水线运行期间自动加载新模型文件命名唯一性每次更新必须上传全新文件名的模型文件同名覆盖不会被识别为更新初始模型必须存在流水线启动前file_pattern至少要匹配到一个文件否则流水线直接失败仅限流式模式WatchFilePattern产生无界数据源只能在 streaming runner如 Dataflow、Flink、Samza上运行更新时机不确定侧输入刷新频率取决于 runner不要假设上传后立即生效可用interval控制轮询间隔版本可追溯通过PredictionResult.model_id与model_name前缀的指标可以在产出与监控两个维度追踪当前生效的模型版本框架无关该模式适用于 TensorFlow、PyTorch 等所有实现了ModelHandler.update_model_path的框架处理器也支持KeyedModelHandler多模型场景。如需深入可继续阅读仓库中的 base.pyRunInference与ModelMetadata实现、utils.pyWatchFilePattern实现、utils_test.py内部逻辑单元测试以及官方文档 about-ml.md 中关于自动模型刷新的章节。赞分享【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载相关推荐Apache Beam 实战使用 WatchFilePattern 作为侧输入自动热更新 RunInference 中的 ML 模型Apache Beam 实战使用 WatchFilePattern 作为侧输入自动热更新 RunInference 中的 ML 模型 Apache Beam大数据批处理流处理数据工程Apache Beam 模型热更新实战用 RunInference 与 WatchFilePattern 侧输入实现 ML 模型自动刷新Apache Beam 模型热更新实战用 RunInference 与 WatchFilePattern 侧输入实现 ML 模型自动刷新 导读 生产环境中的机批处理流处理大数据Apache Beam 模型热更新实战利用 RunInference 与 WatchFilePattern 侧输入实现 ML 模型自动刷新Apache Beam 模型热更新实战利用 RunInference 与 WatchFilePattern 侧输入实现 ML 模型自动刷新 导读在生产 ML大数据批处理流处理数据工程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
