Apache Beam 2.46.0 版本深度解析：RunInference 模型服务增强、I/O 能力扩展与破坏性变更指南

Apache Beam 2.46.0 发布于 2023 年 3 月 10 日，是 Beam 统一批流处理编程模型在机器学习推理（RunInference）、I/O 生态与运行环境治理方面的一次重要迭代。本文以官方发布说明 beam-2.46.0.md 为骨架，结合当前仓库中的 Python/Java SDK 源码，系统梳理本版本的四大看点：Java 容器基础镜像迁移至 Eclipse Temurin、Python SDK RunInference 全面支持模型路径侧输入与 ONNX/TensorFlow 推理、全新的顶层 CSV/JSON 读写 Transform，以及 Spark 2 支持的正式移除。读者读完本文，将能准确评估升级影响、掌握 RunInference 动态模型更新与批量元素（BatchElements）行为调整的配置方法，并理解 Avro 模块迁移的完整路径。

版本概览：2.46.0 的核心主题

2.46.0 是 Beam 在 2023 年开年的首个正式版本，官方发布说明将其定位为"同时包含改进与新功能"（improvements and new functionality）。从变更内容看，本版本的技术主线集中在三个方面：

• ML 推理产品化：RunInference PTransform 获得模型路径侧输入（SideInput）支持，新增 ONNX Runtime 后端与 TensorFlow Model Handler，模型热更新能力通过 WatchFilePattern 得到落地；
• 数据读取体验统一：Python SDK 新增顶层 ReadFromCsv / WriteToCsv / ReadFromJson / WriteToJson Transform，并对文本文件加入 LZMA 压缩编解码支持；
• 运行环境与依赖治理：Java SDK 容器从已弃用的 OpenJDK 基础镜像迁移到 Eclipse Temurin，Java Avro 类从 beam-sdks-java-core 迁往独立的 beam-sdks-java-extensions-avro 模块，Go SDK 构建要求提升至 Go 1.19。

需要特别提醒的是，本版本同时包含若干破坏性变更：Spark 2 的 SparkRunner 被移除，Python BatchElements 的默认批处理策略更激进（目标批时长从 1 秒提升到 10 秒）。计划升级到 2.46.0 的团队应优先评估这两处行为变化。

核心亮点一：Java SDK 容器迁移至 Eclipse Temurin

2.46.0 将 Java SDK 的容器基础镜像从已弃用的 OpenJDK 迁移到 Eclipse Temurin。这是一次由上游生态驱动的必要治理动作：OpenJDK 官方 Docker 镜像已停止维护，而 Temurin 是 Eclipse Adoptium 项目维护的高质量 OpenJDK 发行版。

迁移带来的直接变化是基础操作系统的切换：

• Eclipse Temurin 镜像基于 Ubuntu 22.04；
• 此前的 OpenJDK 镜像基于 Debian 11。

对用户而言，这意味着运行在 Beam Java SDK 容器中的 Pipeline（尤其是 Dataflow、Flink 等基于容器的 Runner）将自动获得更新的 glibc 与系统库，但同时也意味着任何依赖 Debian 专有包或旧系统库的 Pipeline 需要重新验证兼容性。升级后建议对自定义镜像或依赖系统级动态库的 UDF 做一轮回归测试。

核心亮点二：RunInference 机器学习推理能力全面增强

2.46.0 对 Python SDK 的 RunInference PTransform 做了大量投入，使其成为本版本最具技术含量的亮点。RunInference 用于在 Beam Pipeline 中对无界/有界数据流执行模型推理，其核心抽象是 ModelHandler（模型加载与推理逻辑的载体）与 KeyedModelHandler（按 Key 分模型的处理器）。

2.1 模型路径可作为 SideInput（Issue 24042）

本版本之前，模型路径通常作为 Pipeline 构造期的静态参数传入 ModelHandler。2.46.0 起，RunInference 允许将模型路径作为 SideInput 动态传入，配合文件模式监听，即可在 Pipeline 运行期间完成模型热更新，而无需重启作业。

实现这一能力的核心是新增的 WatchFilePattern Transform，其源码位于 sdks/python/apache_beam/ml/inference/utils.py：

class WatchFilePattern(beam.PTransform):
  def __init__(
      self,
      file_pattern,
      interval=360,
      stop_timestamp=MAX_TIMESTAMP,
  ):

从源码可以确认其关键行为：

• 参数语义：file_pattern 支持本地路径与 GCS gs:// 路径，可含 glob 通配符（*、?、[...]）；interval 为轮询检查间隔（秒），默认 360 秒；stop_timestamp 为停止检查的时间戳；
• 实现链路：expand 内部依次串联 MatchContinuously（产生无界 PCollection）、_GetLatestFileByTimeStamp（依据文件修改时间筛选出 Pipeline 启动后新增/更新的文件，并结合 state 记录已见最大时间戳）、_ConvertIterToSingleton（收敛为单元素侧输入）与全局窗口+处理时间触发器的重触发窗口化，从而输出 ModelMetadata 供 RunInference 消费；
• 三条硬性约束（源码 docstring 中明确给出）：
  1. 文件名不可复用——若某文件名曾被使用，后续即使更新内容也会被忽略，触发模型更新必须上传新文件名的文件；
  2. Pipeline 启动前，file_pattern 必须能匹配到至少一个已存在的文件；
  3. 该 Transform 依赖 MatchContinuously 的无界数据源，仅适用于流式模式，批处理模式下可能导致 Pipeline 卡住或产生非预期结果。

配套的单元测试位于 sdks/python/apache_beam/ml/inference/utils_test.py，覆盖了 WatchFilePatternTest 的核心行为。

2.2 ONNX Runtime 后端（Issue 22972）

2.46.0 为 Python SDK 的 RunInference 新增 ONNX Runtime 支持，对应实现文件为 sdks/python/apache_beam/ml/inference/onnx_inference.py，并配套集成测试 onnx_inference_it_test.py 与单元测试 onnx_inference_test.py。这意味着用户可以直接把 .onnx 格式模型接入 Beam Pipeline，借助 ONNX Runtime 在多种硬件后端（CPU/GPU 等）上执行推理，从而在 Beam 内部完成从数据预处理到模型打分的全链路。

2.3 TensorFlow Model Handler（Issue 25366）

同一版本还引入了 TensorFlow 模型的 ModelHandler 支持，使 TensorFlow SavedModel 等格式可以通过 RunInference 加载推理。这与 ONNX、PyTorch 处理器共同构成 2.46.0 的多框架推理矩阵。

2.4 PyTorch：TorchScript 模型加载（PR 25321）

在 PyTorch 侧，2.46.0 新增了对 TorchScript 模型的加载支持。当前仓库中的实现位于 sdks/python/apache_beam/ml/inference/pytorch_inference.py（PytorchModelHandlerTensor 等类），其构造参数中包含 torch_script_model_path，用法如下：

PytorchModelHandlerTensor(
    torch_script_model_path="gs://your-bucket/model.pt")

源码中同时给出了两条约束（见 pytorch_inference.py）：torch_script_model_path 与 state_dict_path 只能二选一，二者同时指定会抛出 ValueError；该路径既支持本地文件也支持 GCS 等 Beam FileSystems 可访问的 URI（源码通过 FileSystems.open 读取模型）。

2.5 实战组合：模型热更新 Pipeline

综合以上能力，一个典型的"流式推理 + 模型热更新"Pipeline 可以这样组织：

import apache_beam as beam
from apache_beam.ml.inference.base import RunInference
from apache_beam.ml.inference.utils import WatchFilePattern
from apache_beam.ml.inference.pytorch_inference import PytorchModelHandlerTensor

model_handler = PytorchModelHandlerTensor(torch_script_model_path="gs://bucket/models/init.pt")

with beam.Pipeline() as p:
    model_updates = (
        p
        | WatchFilePattern(file_pattern="gs://bucket/models/*.pt", interval=60)
    )
    _ = (
        p
        | "ReadData" >> beam.io.ReadFromPubSub(topic="input")
        | "RunInference" >> RunInference(model_handler, model_path=model_updates)
        | "WriteResult" >> beam.io.WriteToBigQuery("dataset.predictions")
    )

其中 WatchFilePattern 每 60 秒轮询一次模型目录，一旦出现文件名全新的 .pt 模型即作为 SideInput 推送给 RunInference，实现不中断作业的模型迭代。请务必遵守前文提到的"唯一文件名"约束与"流式模式"前提。

I/O 增强：JmsIO 重试、LZMA 压缩与顶层 CSV/JSON Transform

3.1 JmsIO 发布重试策略（Java，Issue 24971）

2.46.0 为 Java SDK 的 JmsIO（JMS 消息队列 I/O）新增了对发布失败（failed publications）的重试策略支持。这对于以 JMS 为中间层的生产型 Pipeline 意义重大：此前单次发布失败会直接导致元素处理失败，现在可以配置重试以吸收瞬时故障。JmsIO 实现位于 sdks/java/io 目录下，生产环境使用时应结合目标 MQ 的可用性合理设置重试次数与退避参数。

3.2 文本文件 LZMA 压缩/解压缩（Python，Issue 25316）

Python SDK 的文本 I/O 新增 LZMA 压缩格式支持。从源码看，压缩类型注册表位于 sdks/python/apache_beam/io/filesystem.py，其中定义了 LZMA = 'lzma' 类型、MIME 类型 application/lzma 与 .lzma 扩展名映射；底层编解码由 Python 标准库 lzma.LZMADecompressor / lzma.LZMACompressor 完成（见同文件 第 198、212 行），测试覆盖见 filesystem_test.py。这意味着 ReadFromText / WriteToText 现在可以直接读写 .lzma 后缀文件（如：beam.io.ReadFromText("gs://bucket/data/*.lzma")），无需用户手工解压。

3.3 顶层 ReadFrom/WriteTo Csv/Json Transform（Python）

2.46.0 将 CSV/JSON 读写提升为 Python SDK 的顶层 Transform，源码位于 sdks/python/apache_beam/io/textio.py，内部委托给 apache_beam.dataframe.io 的 ReadViaPandas / WriteViaPandas 实现，即底层使用 pandas 进行解析与序列化。

四个 Transform 的关键签名与参数（均来自源码 docstring）：

Transform	核心参数	说明
ReadFromCsv(path, splittable=True, filename_column=None, **kwargs)	splittable：CSV 是否可按行拆分（单条记录跨多行时应设为 False，会禁用 liquid sharding）；filename_column：追加记录来源文件名的新列名	其余 kwargs 透传 pandas.read_csv
WriteToCsv(path, num_shards=None, file_naming=None, **kwargs)	num_shards：分片数，默认 None 由系统决定；file_naming：文件命名策略，默认 path-XXXXX-of-NNNNN	其余 kwargs 透传 pandas.DataFrame.to_csv
ReadFromJson(path, orient='records', lines=True, dtype=False, **kwargs)	orient：JSON 元素格式，默认 records（对象列表）；lines：是否逐行一条记录，默认 True；dtype：默认 False 不推断类型	其余 kwargs 透传 pandas.read_json
WriteToJson(path, num_shards=None, file_naming=None, orient='records', lines=None, **kwargs)	lines 默认随 orient == 'records' 自动取 True	其余 kwargs 透传 pandas.DataFrame.to_json

一个完整的读写示例：

import apache_beam as beam

with beam.Pipeline() as p:
    records = (
        p
        | beam.io.ReadFromCsv("gs://bucket/input/*.csv", filename_column="source_file")
    )
    _ = (
        records
        | beam.io.WriteToJson("gs://bucket/output/result", num_shards=4)
    )

注意：这些 Transform 依赖 pandas，若未安装 apache_beam<a href="https://link.gitcode.com/i/68d5451bffa5162dc19034f274a707c0" target="_blank">dataframe] 扩展，textio.py 会在导入时抛出 ImportError（源码 [第 1110-1117 行 的 fallback 分支）。单元测试见 textio_test.py，覆盖了 CSV 写入后回读的完整闭环。

新特性与改进：Runner 与多语言 SDK

4.1 Samza 便携模式 UDF 指标

2.46.0 为 Samza Runner 的便携（portable）模式新增 UDF 指标支持，使运行在 Samza 上的跨语言 Pipeline 也能上报并聚合用户自定义指标，补齐了 Samza 便携模式在可观测性方面的缺口。

4.2 SparkRunner：SDF 输出无需全部驻留内存（Issue 23852）

本版本为 SparkRunner 新增实验性选项 use_bounded_concurrent_output_for_sdf，用于避免 SDF（Splittable DoFn）输出必须全部装入内存的限制。开启方式是在 Pipeline 选项的 experiments 中加入该实验项：

PipelineOptions options = PipelineOptionsFactory.create();
options.as(SparkRunnerOptions.class).setExperiments(
    Arrays.asList("use_bounded_concurrent_output_for_sdf"));

这对 ParquetIO 读取等产生大量中间输出的场景尤其有帮助，能以有界并发输出的方式降低内存压力。

4.3 Go SDK：Go 1.19 构建要求与 Prism Runner（PR 25545 / 24789）

Go SDK 在本版本有两项基础性变化：

• 构建要求提升：Go SDK 现在要求 Go 1.19 及以上版本才能编译（见 sdks/go/README.md 与 prepare_go_version.sh 等脚本所管理的版本工具链）；
• Prism Runner 初版落地：Go SDK 新增了首个纯 Go 实现的便携 Beam Runner——Prism，其实现位于 sdks/go/pkg/beam/runners/prism。作为"原生 Go 的可移植 Runner"，Prism 为在无 JVM 环境中运行 Beam 管道提供了轻量选项，适合本地开发调试与对部署体积敏感的场景（当前仓库 runners/prism/java 下亦有配套的 Java 侧组件）。

破坏性变更：升级前必须知晓的两处行为变化

5.1 Spark 2 的 SparkRunner 被移除（PR 25263）

自 2.41.0 起标记弃用的 Spark 2 版本 SparkRunner 在 2.46.0 被正式移除。仍在 Spark 2 集群上运行的团队必须升级到 Spark 3 并使用对应版本的 SparkRunner，否则无法继续使用 Beam Spark 运行后端。当前仓库的 Spark 集成代码位于 runners/spark 目录（含 Spark 3/4 版本子目录），可作为迁移后的对照参考。

5.2 Python BatchElements 批处理行为调整

BatchElements（用于将元素按批聚合以摊销下游固定开销的 Transform）在 2.46.0 中调整了批大小估计策略，其 Python 实现位于 sdks/python/apache_beam/transforms/util.py。具体变化：

• 默认目标批时长从 1 秒提升到 10 秒（源码构造参数默认值 target_batch_duration_secs=10，见 util.py 第 1076 行）；
• 固定开销（fixed cost）不再计入该时长计算，以更好地处理固定开销大于单秒的场景，避免批大小被单元素固定成本严重压低。

行为影响：某些场景下批处理会"更激进"，即单批聚合更多元素、吞吐更高，但单个元素的端到端延迟可能增大。若希望恢复旧行为，显式传入：

from apache_beam.transforms.util import BatchElements

pcoll | BatchElements(target_batch_duration_secs_including_fixed_cost=1)

target_batch_duration_secs_including_fixed_cost 的语义为"含固定开销的每批目标总时长（秒）"，源码会对非正值参数抛出 ValueError（见 util.py 第 532-543 行），其批大小估计逻辑按 fixed_cost + num_elements * per_element_cost 的成本模型线性插值（util.py 第 710-715 行）。相关测试见 util_test.py。

弃用与迁移指南：Avro 模块迁移（Issue 24749）

2.46.0 宣布 beam-sdks-java-core 中的 Avro 相关类正式弃用，并将在未来版本移除。迁移目标是新的独立模块 beam-sdks-java-extensions-avro：

• 包路径迁移：例如 org.apache.beam.sdk.coders.AvroCoder 应改为 org.apache.beam.sdk.extensions.avro.coders.AvroCoder；
• 兼容性保证：为降低迁移成本，新模块中 Avro 类的相对包路径与整体类层级保持不变——只是把包前缀从 org.apache.beam.sdk 换成 org.apache.beam.sdk.extensions.avro，其余结构一一对应；
• 落地验证：当前仓库中迁移后的类已存在，例如 AvroCoder.java。

迁移方式以 Gradle 项目为例，将依赖从 core 模块切换到 extensions 模块，并把 import 语句批量替换为 org.apache.beam.sdk.extensions.avro.* 即可。建议在升级到 2.46.0 时同步完成该迁移，避免未来版本移除旧类时阻塞升级。

贡献者与发布信息

2.46.0 的发布得到了社区广泛参与。根据 git shortlog 统计，本版本共有 70 余位贡献者参与，涵盖核心维护者（如 Danny McCormick、Robert Burke、Luke Cwik、Reuven Lax 等）与众多社区新面孔。发布下载信息可参阅官网下载页的 2.46.0 条目（2023-03-10），更详细的 issue 级变更清单可在官方发布里程碑中查阅。

总结：2.46.0 的升级评估清单

维度	变化	行动项
Java 容器基础镜像	OpenJDK(Debian 11) → Eclipse Temurin(Ubuntu 22.04)	回归测试依赖系统库的 UDF
RunInference	模型路径 SideInput、ONNX、TensorFlow、TorchScript	流式场景可实施模型热更新
Python 顶层 I/O	ReadFrom/WriteTo Csv/Json、LZMA 文本压缩	安装 apache_beam[dataframe] 后方可使用 CSV/JSON Transform
SparkRunner	Spark 2 移除；新增 use_bounded_concurrent_output_for_sdf 实验	Spark 2 集群需先迁移至 Spark 3
BatchElements	默认目标批时长 1s → 10s，固定开销不计入	需旧行为时传 target_batch_duration_secs_including_fixed_cost=1
Go SDK	构建要求 Go 1.19；Prism Runner 初版	升级本地 Go 工具链
Avro	core 模块弃用，迁往 extensions-avro	批量替换 import 包前缀

总体而言，2.46.0 是 Beam 在"ML 推理产品化"方向上迈出的一大步，同时通过容器治理与模块重构为长期可维护性奠基。建议所有用户重点验证 RunInference 相关新特性（尤其是 WatchFilePattern 的流式约束）与 BatchElements 行为变化对现有 Pipeline 吞吐/延迟的影响。