【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载本文以 Apache Beam 官方博客《New Resources Available for Beam ML》为线索系统梳理 Beam 在机器学习领域沉淀的文档资源、核心变换RunInference与示例 Notebook 生态。读者将掌握 Beam 处理 AI/ML 工作负载的完整路径从文档着陆页导航、RunInference 变换的底层原理动态批处理、模型共享、多模型、死信队列到 20 余个可直接运行的 Jupyter Notebook 的选用与上手方法并了解社区后续规划的技术方向。一、背景Beam 在 ML 领域的快速演进在过去一年中Apache Beam 持续发布面向机器学习的功能特性其中最具代表性的是两点RunInference变换的引入一个专门为机器学习推理优化的PTransform让用户可以在批处理与流式管道中直接调用 ML 模型。Beam Dataframes的持续打磨提供类似 Pandas 的 DataFrame API简化数据处理与预处理流程。虽然功能开发进展迅速但此前一直缺乏系统性的资源来帮助用户发现和使用这些新特性。为此Beam 社区投入数月时间补齐文档与 Notebook形成了全新的Beam ML 体验以 AI/ML 着陆页为入口配合一套覆盖数据预处理、推理、模型训练与评估的示例 Notebook。二、入口新的 AI/ML 文档着陆页Beam 官方博客推荐的首个落脚点是新的 AI/ML 着陆页。该页面按照“我想用 Beam ML 做什么”组织内容覆盖了五大工作负载方向方向说明对应文档预测与推理Prediction and inference在批/流管道中本地或远程运行 ML 模型About Beam ML数据处理Data processing数据校验、预处理、探索Preprocess data with MLTransform、Data exploration workflow工作流编排Workflow orchestration与 Kubeflow PipelinesKFP、TensorFlow ExtendedTFX集成Workflow Orchestration模型训练Model training每实体训练、文本聚类、性能基准Per Entity Training、Online Clustering、ML Model Evaluation异常检测Use cases端到端业务场景Anomaly Detection Example页面还整理了丰富的推理专项文档包括多模型管道、带指标的推理RunInference metrics、在线聚类训练、大语言模型推理、TensorRT 自定义模型处理器、跨语言Java SDK 调用推理以及自动模型刷新side input 方式。从源码结构看Beam 的 ML 能力模块集中在 sdks/python/apache_beam/ml/ 目录inference/存放各框架的推理实现transforms/存放 MLTransform 等数据处理变换gcp/存放与 Google Cloud 集成的相关逻辑。文档、实现与 Notebook 三者形成了完整闭环。三、核心变换RunInference 的底层原理RunInference是这一切资源的中心。它自 Apache Beam 2.40.0 起在 Python SDK 中可用2.41.0 起可通过多语言管道框架Multi-language Pipelines从 Java SDK 调用。3.1 核心抽象ModelHandler 与 PredictionResultRunInference的必备参数是一个ModelHandler实现base.py 源码。ModelHandler是通用泛型类职责抽象为两个核心方法load_model()加载并初始化模型run_inference(batch, model, inference_args)对一批样本执行推理并返回结果。输出是PredictionResultbase.py 源码——一个包含example输入样本与inference推理结果的 NamedTuple保留输入便于后续定位预测来源。若输入数据带 key输出则为Tuple[key, PredictionResult]。管道中只需三行即可接入推理from apache_beam.ml.inference.base import RunInference with pipeline as p: predictions ( p | Read beam.ReadFromSource(a_source) | RunInference RunInference(model_handler) )3.2 动态批处理与模型共享RunInference的两个关键优化点expand 实现分别是动态批处理在推理前插入beam.BatchElements变换根据管道吞吐量动态聚合元素以利用向量化推理可通过在ModelHandler中重写batch_elements_kwargs()调整min_batch_size/max_batch_size等参数。模型共享借助Shared/multi_process_shared机制让每个进程只加载一次模型并在该进程内的所有 DoFn 实例间共享显著降低内存占用与加载耗时。当处理多模型时_ModelManagerbase.py 源码按 LRU 策略管理各模型在内存中的生命周期支持max_models_per_worker_hint限制同时驻留的模型数量避免多模型并发加载导致 OOM。3.3 开箱即用的框架支持仓库 sdks/python/apache_beam/ml/inference/ 中可以看到官方预置的模型处理器覆盖PyTorchPytorchModelHandlerTensor、PytorchModelHandlerKeyedTensor等pytorch_inference.pyScikit-learnSklearnModelHandlerNumpy、SklearnModelHandlerPandas支持ModelFileType.PICKLE/ModelFileType.JOBLIBsklearn_inference.pyTensorFlowTFModelHandlerNumpy/TFModelHandlerTensor亦可通过tfx-bsl的CreateModelHandler处理tf.Example输入tensorflow_inference.py以及XGBoost、Hugging Face、Vertex AI、TensorRT、ONNX等对应xgboost_inference.py、huggingface_inference.py、vertex_ai_inference.py、tensorrt_inference.py、onnx_inference.py。如果框架不在列表中可自定义ModelHandler实现load_model与run_inference参考 run_custom_inference.ipynb 中以 spaCy 为例的写法。3.4 面向生产的模式多模型管道支持 A/B 模式同一数据分别过模型 A、B与级联模式前一模型输出经后处理喂给后一模型并可配合with_resource_hints为不同推理步骤声明 RAM 或加速器需求按 key 路由模型KeyedModelHandlerKeyModelMapping可按输入 key 选择不同模型适合多实体多模型的场景自动模型刷新通过WatchFilePattern作为 side input 传入model_metadata_pcoll在不停止管道的情况下热更新模型详见 side-input-updates.md死信队列调用with_exception_handling()后返回RunInferenceDLQ将失败的预处理、推理、后处理记录分别输出到独立 PCollection 供重试或分析base.py 源码。注意RunInference采用动态批处理但无法对不同尺寸的张量做 stack因此输入样本需保持相同维度对文本等变长输入可在ModelHandler中重写batch_elements_kwargs设置max_batch_size1来关闭批处理。四、示例 Notebook 库Beam ML 实战图谱博客配套推出了位于 examples/notebooks/beam-ml/ 的 Jupyter Notebook 集合全部围绕RunInference与Beam Dataframes展开。按其 README 说明可按场景分类数据预处理data_preprocessing/子目录下的 Vertex AI / Hugging Face 文本嵌入生成、词表计算与应用、MLTransform 数据缩放以及dataframe_api_preprocessing.ipynb展示 DataFrame API 预处理。预训练模型推理覆盖 PyTorchrun_inference_pytorch.ipynb、scikit-learnrun_inference_sklearn.ipynb、TensorFlowrun_inference_tensorflow.ipynb、TensorFlow Hub、Hugging Face、XGBoost、TFX、Vertex AI 远程端点、多框架综合run_inference_pytorch_tensorflow_sklearn.ipynb、大语言模型run_inference_generative_ai.ipynb以及带窗口语义的推理run_inference_windowing.ipynb。自定义推理run_custom_inference.ipynb自带模型与custom_remote_inference.ipynb远程 API 推理。业务用例image_processing_tensorflow.ipynb图像处理、nlp_tensorflow_streaming.ipynb流式 NLP、speech_emotion_tensorflow.ipynb语音情感识别。自动模型刷新automatic_model_refresh.ipynb演示生产环境热更新模型。多模型管道run_inference_multi_model.ipynb以图像字幕生成 排序的级联集成模型为例per_key_models.ipynb演示多个不同训练模型的按 key 路由推理。模型评估tfma_beam.ipynb使用 TensorFlow Model AnalysisTFMA在管道内评估与对比模型性能。4.1 如何运行 Notebook按 examples/notebooks/beam-ml/README.md 的指引使用 Colab 等支持 Jupyter Notebook 的环境在环境中打开仓库内对应 Notebook 文件运行大多数 Notebook 前需将其中配置的 Google Cloud 项目与存储桶替换为你自己的 project 和 bucket。五、后续规划与社区参与博客明确指出扩充示例与 Notebook 是 Beam ML 后续迭代的重点。下一轮改进计划包括在 RunInference 中支持超过 30GB 的大模型支持多语言管道跨 SDK 调用推理与常见 Beam 概念窗口、触发器等结合的示例TensorRT集成示例覆盖机器学习生命周期其他环节如TFMA 模型评估、每实体训练per-entity training与更多在线训练示例。部分规划已在仓库中落地成型tensorrt_inference.py与 tensorrt-runinference.md 对应 TensorRT 支持tfma_beam.ipynb对应模型评估per_key_models.ipynb与 per-entity-training.md 对应每实体训练run_inference_generative_ai.ipynb对应 LLM 推理。社区欢迎以 issue 或 pull request 的方式反馈改进建议。若希望进一步深入可继续阅读 About Beam ML含完整配置参数与故障排查、RunInference 指标文档并参考 sdks/python/apache_beam/examples/inference/ 下的完整管道示例。结语Apache Beam 的 ML 资源体系已从“功能先行”走向“资源配套”以 AI/ML 着陆页为导航、以 RunInference 为核心变换、以 20 余个 Notebook 为实战样例覆盖预处理、推理、训练、评估与编排的完整 AI/ML 生命周期。无论你是要在一个已有管道中接入预训练模型还是从零搭建多模型推理与自动刷新的生产链路这套资源都能提供直接可用的起点。赞分享【免费下载链接】beamApache Beam is a unified programming model for Batch and Streaming data processing.项目地址https://gitcode.com/gh_mirrors/beam18/beam点击查看免费下载相关推荐Apache Beam ML 资源指南RunInference 与示例 Notebook 全景解析Apache Beam ML 资源指南RunInference 与示例 Notebook 全景解析 导读 本文以 Apache Beam 官方博客《New大数据批处理流处理数据工程Apache Beam ML Notebooks 全览从 RunInference 到 MLTransform 的实战指南Apache Beam ML Notebooks 全览从 RunInference 到 MLTransform 的实战指南 本文以 Apache Beam 仓批处理流处理大数据Apache Beam Beam ML 示例 Notebook 体系RunInference、ModelHandler 与 MLTransform 的机器学习实战Apache Beam Beam ML 示例 Notebook 体系RunInference、ModelHandler 与 MLTransform 的机器学习大数据批处理流处理数据工程上一篇3分钟搞定PotPlayer字幕翻译插件终极配置指南让外语视频无障碍观看下一篇GetQzonehistory一次扫码导出 QQ 空间全部历史说说创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考