根据一个或多个贡献者许可协议，许可给Apache软件基金会（ASF）。

查看NOTICE文件以获取有关版权所有权的额外信息。

ASF根据Apache许可证版本2.0（“许可证”）许可此文件；

除非符合许可证，否则不得使用此文件。您可以在

http://www.apache.org/licenses/LICENSE-2.0

获取许可证的副本。

除非适用法律要求或书面同意，软件根据许可证按“原样”分发，

没有任何明示或暗示的保证或条件。请参阅许可证以了解

特定语言下的权限和限制。

name: beam-concepts description: 解释Apache Beam编程模型的核心概念，包括PCollections、PTransforms、Pipelines和Runners。在学习Beam基础或解释管道概念时使用。

Apache Beam核心概念

Beam模型

从Google的MapReduce、FlumeJava和Millwheel项目演化而来。最初称为“数据流模型”。

关键抽象

管道

管道封装了整个数据处理任务，包括读取、转换和写入数据。

// Java
Pipeline p = Pipeline.create(options);
p.apply(...)
 .apply(...)
 .apply(...);
p.run().waitUntilFinish();

# Python
with beam.Pipeline(options=options) as p:
    (p | 'Read' >> beam.io.ReadFromText('input.txt')
       | 'Transform' >> beam.Map(process)
       | 'Write' >> beam.io.WriteToText('output'))

PCollection

一个分布式数据集，可以是有界的（批处理）或无界的（流处理）。

属性

不可变 - 一旦创建，无法修改
分布式 - 元素并行处理
可能有界或无界
带时间戳 - 每个元素都有事件时间戳
窗口化 - 元素分配到窗口

PTransform

一个数据处理操作，用于转换PCollections。

// Java
PCollection<String> output = input.apply(MyTransform.create());

# Python
output = input | 'Name' >> beam.ParDo(MyDoFn())

核心转换

ParDo

通用并行处理。

// Java
input.apply(ParDo.of(new DoFn<String, Integer>() {
    @ProcessElement
    public void processElement(@Element String element, OutputReceiver<Integer> out) {
        out.output(element.length());
    }
}));

# Python
class LengthFn(beam.DoFn):
    def process(self, element):
        yield len(element)

input | beam.ParDo(LengthFn())
# 或更简单：
input | beam.Map(len)

GroupByKey

按键分组元素。

PCollection<KV<String, Integer>> input = ...;
PCollection<KV<String, Iterable<Integer>>> grouped = input.apply(GroupByKey.create());

CoGroupByKey

按键连接多个PCollections。

Combine

组合元素（求和、平均值等）。

// 全局组合
input.apply(Combine.globally(Sum.ofIntegers()));

// 按键组合
input.apply(Combine.perKey(Sum.ofIntegers()));

Flatten

合并多个PCollections。

PCollectionList<String> collections = PCollectionList.of(pc1).and(pc2).and(pc3);
PCollection<String> merged = collections.apply(Flatten.pCollections());

Partition

将PCollection拆分为多个PCollections。

窗口化

类型

固定窗口 - 规则、非重叠间隔
滑动窗口 - 重叠间隔
会话窗口 - 不活动间隙定义边界
全局窗口 - 所有元素在一个窗口中（默认）

input.apply(Window.into(FixedWindows.of(Duration.standardMinutes(5))));

input | beam.WindowInto(beam.window.FixedWindows(300))

触发器

控制结果何时发射。

input.apply(Window.<T>into(FixedWindows.of(Duration.standardMinutes(5)))
    .triggering(AfterWatermark.pastEndOfWindow()
        .withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
            .plusDelayOf(Duration.standardMinutes(1))))
    .withAllowedLateness(Duration.standardHours(1))
    .accumulatingFiredPanes());

侧输入

ParDo的附加输入。

PCollectionView<Map<String, String>> sideInput =
    lookupTable.apply(View.asMap());

mainInput.apply(ParDo.of(new DoFn<String, String>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
        Map<String, String> lookup = c.sideInput(sideInput);
        // 使用查找...
    }
}).withSideInputs(sideInput));

管道选项

配置管道执行。

public interface MyOptions extends PipelineOptions {
    @Description("输入文件")
    @Required
    String getInput();
    void setInput(String value);
}

MyOptions options = PipelineOptionsFactory.fromArgs(args).as(MyOptions.class);

模式

结构化数据的强类型访问。

@DefaultSchema(AutoValueSchema.class)
@AutoValue
public abstract class User {
    public abstract String getName();
    public abstract int getAge();
}

PCollection<User> users = ...;
PCollection<Row> rows = users.apply(Convert.toRows());

错误处理

死信队列模式

TupleTag<String> successTag = new TupleTag<>() {};
TupleTag<String> failureTag = new TupleTag<>() {};

PCollectionTuple results = input.apply(ParDo.of(new DoFn<String, String>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
        try {
            c.output(process(c.element()));
        } catch (Exception e) {
            c.output(failureTag, c.element());
        }
    }
}).withOutputTags(successTag, TupleTagList.of(failureTag)));

results.get(successTag).apply(WriteToSuccess());
results.get(failureTag).apply(WriteToDeadLetter());

跨语言管道

使用其他SDK中的转换。

# 从Python使用Java Kafka连接器
from apache_beam.io.kafka import ReadFromKafka

result = pipeline | ReadFromKafka(
    consumer_config={'bootstrap.servers': 'localhost:9092'},
    topics=['my-topic']
)

最佳实践

优先使用内置转换而非自定义DoFns
使用模式进行类型安全操作
最小化侧输入以提高性能
显式处理延迟数据
使用DirectRunner测试后再部署
使用TestPipeline进行单元测试