- 批处理
- 流处理
- 大数据
【免费下载链接】beam
Apache Beam is a unified programming model for Batch and Streaming data processing.
导读
本文围绕 Apache Beam 的 Direct Runner，系统讲解如何在本地机器上运行、测试与调试 Beam 管道。你将了解到 Direct Runner 的设计定位与严格模型校验机制、Java/Python 两种 SDK 下的依赖配置与命令行选项解析方式、核心DirectOptions参数语义，以及该 Runner 在进程内执行管道、串行化用户函数、强制元素不可变与可编码等底层实现原理。读完即可在你的开发机上快速跑通并验证 Beam 管道。
一、Direct Runner 是什么：在进程内执行的轻量级 Runner
Direct Runner 是 Apache Beam 内置的、在构造管道的进程内直接执行整条管道的PipelineRunner。从当前仓库源码可以看到它的类定义与定位声明：
// runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java public class DirectRunner extends PipelineRunner<DirectPipelineResult> { ... }其 Javadoc 明确指出：Direct Runner 适用于小规模、示例和测试数据上的管道运行，用于确保处理逻辑正确，同时适合执行单元测试，并会额外做一些工作，保证管道内的行为不违反 Beam 模型的假设，从而提高管道在分布式后端上大规模运行的能力。
典型适用场景
- 本地开发与快速迭代：改动代码后立即看到运行结果，无需申请集群资源；
- 单元测试与集成测试：直接使用你熟悉的本地调试工具（IDE 断点、日志、
pdb/debugger）即可定位问题； - 逻辑验证：在提交到分布式 Runner（如 Dataflow、Flink、Spark）之前，先用 Direct Runner 验证管道逻辑正确性；
- 教学与示例：本文仓库的 learning/katas 系列练习均基于本地运行方式展开。
需要强调的是，Direct Runner 不是为生产大数据量处理设计的，它不会提供分布式调度、容错、横向扩容等能力，其价值在于"以最严格的方式暴露模型违规问题"。
二、Direct Runner 的模型校验机制：四大强制检查
Direct Runner 相比普通本地执行器最大的不同，是它会对管道执行额外的语义检查，确保用户代码不依赖 Beam 模型未承诺的行为。这些检查包括：
| 检查项 | 含义 | 违反时的表现 |
|---|---|---|
| 元素不可变性（Immutability） | 强制保证PCollection中的元素在处理过程中不被修改；PTransform 在任何时候都不得修改输入元素，也不得在输出元素之后修改它们 | 抛出IllegalMutationException |
| 元素可编码性（Encodability） | 强制保证每个PCollection的所有元素都能被该集合的 Coder 编码与解码 | 抛出 Coder 相关异常 |
| 任意顺序处理 | 确保元素在所有处理点都以任意顺序被处理，不依赖任何顺序保证 | 逻辑依赖顺序的代码可能得到非确定性结果 |
| 用户函数可序列化 | 确保DoFn、CombineFn等用户函数可被序列化 | 序列化失败时抛出异常 |
源码实现印证
在 DirectRunner.java 中定义了Enforcement枚举，其中：
ENCODABILITY：对所有PCollection生效；IMMUTABILITY：对除 Read 变换以外的、包含用户函数（UDF）的变换生效（CONTAINS_UDF集合包含READ_TRANSFORM_URN与PAR_DO_TRANSFORM_URN）。
两个检查开关分别由DirectOptions.isEnforceEncodability()与isEnforceImmutability()控制，默认均为开启。开启时：
- 编码性检查会使用
CloningBundleFactory（对元素做克隆/重编码验证）； - 不可变性检查通过
ImmutabilityCheckingBundleFactory包装 BundleFactory，并在元素进入变换前后用MutationDetector比对。见 ImmutabilityEnforcementFactory.java：它在beforeElement时为每个元素登记MutationDetector，在afterElement/afterFinish时调用verifyUnmodified()，一旦发现IllegalMutationException即抛出带有完整 PTransform 名称与元素类信息的错误。
实践提示：这四大检查意味着"在 Direct Runner 上能通过的管道，换到其他 Runner 也大概率符合模型约束"，这正是 Beam 官方推荐用 Direct Runner 做跨 Runner 兼容性验证的原因。
三、步骤 1：配置依赖（Java）
使用 Java SDK 时，需要在项目的pom.xml中声明对 Direct Runner 的依赖：
<dependency> <groupId>org.apache.beam</groupId> <artifactId>beam-runners-direct-java</artifactId> <version>2.54.0</version> <scope>runtime</scope> </dependency>要点说明：
- scope 使用
runtime：Direct Runner 只在运行期被PipelineRunnerRegistrar动态发现，编译期无需引用，运行时由 DirectRegistrar.java 通过@AutoService机制自动注册DirectRunner与DirectOptions； - 请将
2.54.0替换为你实际使用的 Beam 版本（版本号以你工程依赖的 Beam release 为准）； - 同时确认工程中已包含 Beam SDK 核心依赖（如
beam-sdks-java-core），并把 Java 代码编译为自包含 JAR（可使用mvn package或 GradleshadowJar/fatJar插件），以便后续用java -jar直接运行。
四、步骤 2：配置 Pipeline Options
Beam SDK 内置了命令行参数解析器，可以通过--<option>=<value>的形式从命令行读取配置，无需硬编码在代码里。
Java SDK
PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();withValidation()会在创建时校验必需选项，缺省或非法时直接报错，便于尽早发现问题。若不想用命令行参数，也可以直接构造PipelineOptions对象并调用 setter 设置字段。
Python SDK
from apache_beam.options.pipeline_options import PipelineOptions beam_options = PipelineOptions()Python 侧同样支持从命令行解析，PipelineOptions()默认读取sys.argv。
指定 Runner
在命令行传入：
--runner=DirectRunner或简写别名--runner=direct。其余管道选项一般使用默认值即可；如需更多定制，可参考DirectOptions接口（Java）与apache_beam.options.pipeline_options.DirectOptions（Python）的说明，下文详细介绍。
五、核心配置参数：DirectOptions 详解
Java 侧DirectOptions
定义于 runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java，它继承自PipelineOptions与ApplicationNameOptions，包含以下关键选项：
| 选项 | 默认值 | 说明 |
|---|---|---|
--blockOnRun | true | 调用Pipeline#run()时是否阻塞等待管道完成。为true时run()会阻塞到所有 PTransform 执行完毕；为false时管道异步执行，需调用PipelineResult#waitUntilFinish()等待完成 |
--enforceImmutability | true | 是否强制保证每个PCollection的元素不被修改（见上文不可变性检查） |
--enforceEncodability | true | 是否强制保证每个PCollection的元素都能被其 Coder 编解码（见上文编码性检查） |
--targetParallelism | max(可用处理器数, 3) | Direct Runner 使用的目标并行度，必须大于 0。默认值由内部AvailableParallelismFactory计算：Math.max(Runtime.getRuntime().availableProcessors(), MIN_PARALLELISM)，其中MIN_PARALLELISM = 3（见 DirectOptions.java） |
Python 侧DirectOptions
定义于 sdks/python/apache_beam/options/pipeline_options.py，提供了与 Java 不同的、面向本地执行的细化参数：
| 命令行参数 | 默认值 | 说明 |
|---|---|---|
--direct_running_mode | in_memory | 工作进程运行环境，可选in_memory（纯内存单线程）、multi_threading（多线程）、multi_processing（多进程） |
--direct_num_workers | 1 | 并行运行的 worker 数量 |
--direct_runner_bundle_repeat | 0 | 每个 Bundle 额外重放多少次，用于性能剖析与调试 |
--no_direct_runner_use_stacked_bundle | 开启 stacked bundle | Direct Runner 默认在 Bundle 内使用堆叠（stacked）WindowedValue以优化内存，传入该参数可关闭 |
--direct_embed_docker_python | False | 指定默认 Python Docker 环境时，是否改用嵌入的本地 Python 环境执行 |
--direct_test_splits | {} | 测试用 Split 配置，JSON 格式{"step_name": {"timings": [...], "fractions": [...]}, ...}，用于 SDF（Splittable DoFn）分片测试 |
注意：
--direct_running_mode选择多线程/多进程模式时，需结合--direct_num_workers调整并行度，以获得接近生产 Runner 的并发行为。
六、步骤 3：运行管道
Python SDK
使用python -m语法调用包含管道代码的模块：
python -m your_module_name例如管道定义在wordcount.py中：
python -m wordcount --runner=DirectRunner --input=gs://.../input.txt --output=/tmp/outputJava SDK
运行编译好的 JAR：
java -jar your-project.jar示例：
java -jar wordcount.jar --runner=DirectRunner --inputFile=/path/to/input --output=/tmp/output单元测试方式
Direct Runner 也是单元测试的推荐载体，你可以直接在测试代码中构建Pipeline并run()（阻塞模式），配合PAssert验证输出集合。此时--blockOnRun=true可确保测试方法返回前管道已完整执行。
七、源码级运行原理：管道如何被本地执行
从 DirectRunner.run() 的实现可以还原完整的本地执行链路：
- 选项序列化归一：将
PipelineOptions序列化为 JSON 再反序列化，统一选项表示； - 管道重写（performRewrites）：按固定顺序应用一系列
PTransformOverride，包括：- 侧输入（Side Input）相关重写（
sideInputUsingTransformOverrides，处理 runner 决定的 sharding、多步 Combine 等）； - 为每个活跃侧输入附加
WriteView原语（DirectWriteViewVisitor）； - GroupByKey 相关重写（
groupByKeyOverrides）：Splittable ParDo、带状态/定时器的 ParDo 均被重写为普通 ParDo 加额外原语，GroupByKey被拆分为两个串联原语（DirectRunner.java）；
- 侧输入（Side Input）相关重写（
- 图构建与校验：
DirectGraphVisitor拓扑遍历构建执行图，DisplayDataValidator校验管道与选项； - 上下文与执行器创建：
EvaluationContext持有时钟、BundleFactory 与键控值信息；TransformEvaluatorRegistry注册各变换的求值器；ExecutorServiceParallelExecutor以getTargetParallelism()为并行度启动执行； - 结果对象：返回
DirectPipelineResult，支持getState()查询状态、waitUntilFinish()阻塞等待、cancel()取消，以及metrics()获取指标（DirectRunner.java）。
Python 侧的执行器结构
Python SDK 的 Direct Runner 实现在 sdks/python/apache_beam/runners/direct/direct_runner.py，其中：
SwitchingDirectRunner：默认入口，会根据管道内容自动切换执行方式——批量作业走 FnApiRunner（吞吐高），流式作业或 FnApiRunner 尚未支持的原语（如TestStream、Pub/Sub 读写、带延迟侧输入的 CombineFn）走BundleBasedDirectRunner；BundleBasedDirectRunner：基于 Bundle 的执行模型，支持流式与有状态原语。
这种"自动切换"机制保证了本地行为尽量贴合真实 Runner 的语义覆盖范围。
八、使用注意事项与限制
- 非生产用途：Direct Runner 在单进程内执行，无分布式容错与扩展能力，仅适合小数据量、示例与测试场景；
- 资源与并行度：Java 侧并行度默认取
max(可用 CPU 数, 3)，可通过--targetParallelism调整；Python 侧通过--direct_num_workers与--direct_running_mode控制； - 严格校验可能"误伤"：默认开启的不可变性/可编码性检查会拒绝在分布式 Runner 上"侥幸通过"的代码（例如对输入元素就地修改、使用不可序列化的 Coder），这是有意为之——Direct Runner 应当作为最严格的模型合规性守门员；
- 测试专用隐藏选项：
DirectTestOptions（如--runnerDeterminedSharding，默认true）属于内部测试调优选项（见 DirectTestOptions.java），标注为@Hidden，普通用户不应依赖； - 版本一致性：
pom.xml中的 Direct Runner 依赖版本需与 Beam SDK 版本保持一致，避免运行时版本冲突。
九、总结
Direct Runner 是 Apache Beam 生态中"零部署、零成本、最严格"的本地执行环境：它以进程内执行的方式完整复现 Beam 模型，并通过不可变性、可编码性、任意顺序处理与用户函数序列化四类强制检查，帮你把模型违规问题消灭在开发阶段。掌握依赖配置（Javapom.xml的beam-runners-direct-java）、命令行选项解析（JavaPipelineOptionsFactory.fromArgs(...)、PythonPipelineOptions）、Runner 指定（--runner=DirectRunner）以及DirectOptions系列参数后，你就能在本地快速跑通、调试并验证 Beam 管道，为迁移到分布式 Runner 打下可靠基础。
如需进一步深入，可直接阅读本仓库中的 DirectRunner.java、DirectOptions.java 与 Python 侧实现 direct_runner.py，并结合 learning/katas 中的练习代码上手实操。
- 批处理
- 流处理
- 大数据
【免费下载链接】beam
Apache Beam is a unified programming model for Batch and Streaming data processing.
相关推荐
Apache Beam Direct Runner 本地运行指南：测试、调试与模型语义校验
Apache Beam Direct Runner 本地运行指南：测试、调试与模型语义校验 导读：本文围绕 Apache Beam 的 Direct Runne
大数据批处理流处理数据工程Apache Beam Direct Runner 完全指南：在本地运行与调试 Beam 管道
Apache Beam Direct Runner 完全指南：在本地运行与调试 Beam 管道 导读 Apache Beam 提供了统一的批处理与流处理编程模型
大数据批处理流处理数据工程Apache Beam Direct Runner 完全指南：本地验证 Beam 模型语义的正确性优先执行器
Apache Beam Direct Runner 完全指南：本地验证 Beam 模型语义的正确性优先执行器 Direct Runner 是 Apache Be
大数据批处理流处理数据工程
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
