Apache Beam 技术文档

1. 镜像概述和主要用途

Apache Beam 是一个统一的数据处理模型，用于定义批处理和流处理数据并行处理管道，同时提供一套特定语言的 SDK 用于构建管道，以及在分布式处理后端执行管道的运行器（Runner）。支持的后端包括 Apache Apex、Apache Flink、Apache Spark、Google Cloud Dataflow 和 Hazelcast Jet 等。其核心目标是简化跨平台、跨场景的数据处理逻辑开发，实现"一次编写，多处运行"的能力。

2. 核心功能和特性

2.1 统一批流处理模型

基于 Google Dataflow 模型演进而来，统一批处理和流处理的编程范式，支持无界数据流（流处理）和有界数据流（批处理）的统一表达，简化复杂数据处理逻辑的开发。

2.2 多语言 SDK 支持

提供多种编程语言的 SDK，满足不同技术栈需求：

• Java SDK：成熟稳定，支持完整功能
• Python SDK：适用于数据科学和快速开发场景
• Go SDK：轻量级，适用于云原生环境
• 其他语言支持（如 Scala、R 等）可通过社区扩展

2.3 多运行器兼容

支持在多种分布式处理引擎上执行管道，无需修改业务逻辑：

• DirectRunner：本地执行，用于调试和测试
• ApexRunner：基于 Apache Hadoop YARN 集群执行（支持嵌入式模式）
• DataflowRunner：提交至 Google Cloud Dataflow 服务
• FlinkRunner：在 Apache Flink 集群上执行（代码源自 dataArtisans/flink-dataflow）
• SparkRunner：在 Apache Spark 集群上执行（代码源自 cloudera/spark-dataflow）
• JetRunner：在 Hazelcast Jet 集群上执行（代码源自 hazelcast/hazelcast-jet）

2.4 灵活的管道抽象

核心概念包括：

• PCollection：表示数据集合，支持有界（批处理）或无界（流处理）数据
• PTransform：定义数据转换逻辑，将输入 PCollection 转换为输出 PCollection
• Pipeline：管理 PTransform 和 PCollection 构成的有向无环图，用于执行
• PipelineRunner：指定管道的执行环境和方式

3. 使用场景和适用范围

3.1 终端用户（End Users）

场景：使用现有 SDK 编写数据处理管道，并在指定运行器上执行。
适用范围：数据工程师、数据分析师需快速开发批流处理任务，无需关注底层执行引擎细节。例如：日志数据实时清洗、用户行为分析、ETL 流程构建等。

3.2 SDK 开发者（SDK Writers）

场景：为特定用户群体开发 Beam SDK（如 Scala、R、图形化界面等）。
适用范围：语言专家或框架开发者，需扩展 Beam 的语言生态，满足特定技术栈需求。

3.3 运行器开发者（Runner Writers）

场景：为分布式处理环境开发 PipelineRunner，支持基于 Beam 模型编写的程序。
适用范围：分布式计算框架开发者，需将 Beam 生态集成至自有执行引擎，扩展框架的数据处理能力。

4. 使用方法和配置说明

4.1 Docker 部署方案

4.1.1 基础镜像拉取与运行

Apache Beam 官方未提供统一的 Docker 镜像，建议基于 SDK 语言构建自定义镜像。以下为 Java SDK 示例：

Dockerfile（Java SDK 示例）

FROM maven:3.8.5-openjdk-11 AS builder
WORKDIR /app
COPY pom.xml .
# 缓存依赖
RUN mvn dependency:go-offline
COPY src ./src
# 构建 Beam 管道应用
RUN mvn package -DskipTests

FROM openjdk:11-jre-slim
WORKDIR /app
COPY --from=builder /app/target/*.jar app.jar
# 运行 DirectRunner（本地调试）
ENTRYPOINT ["java", "-jar", "app.jar", "--runner=DirectRunner"]

构建并运行镜像

# 构建镜像
docker build -t beam-java-app:latest .

# 运行（使用 DirectRunner 本地执行）
docker run --rm beam-java-app:latest

4.1.2 分布式运行器配置示例（Flink Runner）

若需提交至 Flink 集群执行，需在运行时指定 Flink 集群地址：

docker run --rm \
  -e FLINK_MASTER=flink-jobmanager:8081 \
  beam-java-app:latest \
  --runner=FlinkRunner \
  --flink-master=${FLINK_MASTER} \
  --streaming=true  # 若为流处理任务

4.1.3 docker-compose 配置（集成 Flink 集群）

version: "3.8"
services:
  flink-jobmanager:
    image: flink:1.17-scala_2.12
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=flink-jobmanager

  flink-taskmanager:
    image: flink:1.17-scala_2.12
    depends_on:
      - flink-jobmanager
    command: taskmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=flink-jobmanager

  beam-app:
    build: .
    depends_on:
      - flink-jobmanager
    environment:
      - FLINK_MASTER=flink-jobmanager:8081
    command: >
      java -jar app.jar
      --runner=FlinkRunner
      --flink-master=${FLINK_MASTER}
      --jobName=beam-flink-demo

4.2 核心配置参数与环境变量

4.2.1 通用参数

4.2.2 运行器特定参数

• FlinkRunner

  参数名	说明	示例值
  --flink-master	Flink 集群 JobManager 地址	flink-jobmanager:8081
  --streaming	是否启用流处理模式	true/false
  --parallelism	任务并行度	4

• SparkRunner

  参数名	说明	示例值
  --spark-master	Spark 集群 Master 地址	spark://spark-master:7077
  --spark-submit	Spark 提交命令路径	/opt/spark/bin/spark-submit

• DataflowRunner

  参数名	说明	示例值
  --project	GCP 项目 ID	my-gcp-project
  --region	区域	us-central1
  --tempLocation	GCS 临时文件路径	gs://my-bucket/temp

4.2.3 环境变量

5. 快速入门示例：WordCount 管道

5.1 Java SDK 示例代码（WordCount.java）

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.FlatMapElements;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.TypeDescriptors;

import java.util.Arrays;

public class WordCount {
  public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
    Pipeline p = Pipeline.create(options);

    p.apply(TextIO.read().from(options.as(WordCountOptions.class).getInputFile()))
     .apply(FlatMapElements.into(TypeDescriptors.strings())
         .via(line -> Arrays.asList(line.split(" "))))
     .apply(Count.perElement())
     .apply(MapElements.into(TypeDescriptors.strings())
         .via(wordCount -> wordCount.getKey() + ": " + wordCount.getValue()))
     .apply(TextIO.write().to(options.as(WordCountOptions.class).getOutput()));

    p.run().waitUntilFinish();
  }

  public interface WordCountOptions extends PipelineOptions {
    String getInputFile();
    void setInputFile(String value);
    String getOutput();
    void setOutput(String value);
  }
}

5.2 构建并运行

1. 将上述代码打包为 JAR（通过 Maven/Gradle）
2. 使用 Docker 运行（DirectRunner 本地模式）：

docker run --rm \
  -v $(pwd)/input.txt:/data/input.txt \
  -v $(pwd)/output:/data/output \
  beam-java-app:latest \
  --runner=DirectRunner \
  --inputFile=/data/input.txt \
  --output=/data/output/result

6. 相关资源与支持

• 官方文档：Apache Beam 官网
• 快速入门：
  • Java SDK
  • Python SDK
  • Go SDK
• 社区支持：
  • 用户邮件列表：***
  • 开发者邮件列表：***
  • issue 跟踪：Apache Beam JIRA