Apache Beam 统一的分布式数据处理编程库
参考文章：
https://www.ctolib.com/topics/106500.html
https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102
https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101
原文：https://blog.csdn.net/u011489205/article/details/79173938
Beam编程指南适用于想要使用Beam SDK创建数据处理管道的Beam用户。 它为使用Beam SDK类构建和测试管道提供了指导。 它不是作为一个详尽的参考，而是作为一个语言无关，高级指南，以编程方式构建您的Beam pipeline。 随着编程指南的填写，文本将包含多种语言的代码示例，以帮助说明如何在管道中实现Beam概念。
目录
总览
创建pipline
Working with PCollections
Creating a PCollection
PCollection Characteristics
Element Type
Immutability
Random Access
Size and Boundedness
Element Timestamps
Applying Transforms
Using ParDo
Using GroupByKey
Using Combine
Using Flatten and Partition
General Requirements for Writing User Code for Beam Transforms
Side Inputs and Side Outputs
Composite Transforms
Pipeline I/O
Running the Pipeline
Data Encoding and Type Safety
Working with Windowing
Working with Triggers
1、概述
为了使用Beam，您需要首先使用其中一个Beam SDK中的类创建一个驱动程序。你的driver程序定义了你的pipeline，包括所有的输入，变换以及输出。他也会设置执行的参数。这些都包含在Pipeline Runner中，确定你的Pipeline 将运行什么后端。
Beam SDK提供了大量简化大规模分布式数据处理机制的抽象。 相同的Beam抽象与批处理和流数据源一起工作。 当你创建你的Beam管道时，你可以考虑这些抽象的数据处理任务。 他们包括：
Pipeline:Pipeline封装了整个数据处理任务，从开始到结束。这个包含读取输入数据，数据的transform，以及数据输出。所有的dirver程序必须创建Pipeline。当你创建Pipeline的时候，必须也要指定执行的参数，来告诉Pipeline在哪里怎样运行。
PCollection:PCollection代表Beam pipline操作的分布式数据集。数据集可以是有界的，意思是来自一个固定的source，例如文件。或者数据集是没有边界的，意味着数据集来自一个持续更新的source通过订阅或者其他手段。pipeline从外部数据源读取数据来创建一个pipeline。但是也可以使用driver程序从内存创建PCollection。PCollection是pipeline中每个步骤的输入输出。
Transform：Transform代表着pipeline中数据处理操作或者步骤。每个Transform任务会有一个或者多个PCollection作为输入，执行您在该PCollection的元素上提供的处理函数，并生成一个或多个输出PCollection对象。
I/OSource和Sink:Beam提供了source和sink APIs来分别表示读写数据，Source封装了从某些外部源将数据读入您的Beam管道所需的代码，例如云文件存储或流式数据源的订阅。 Sink同样封装了将PCollection的元素写入外部数据宿所需的代码。
典型的Beam driver程序工作流程如下：
创建pipeline对象并且设置pipeline执行参数包括pipeline runner；
为pipeline数据创建初始PCollection，使用Source API从外部源读取数据，或使用创建transform从内存数据构建PCollection.
将transform作用到每个PCollection。一个transform操作会创建一个PCollection而不会消费输入集合。典型的pipeline将后续transform依次应用于每个新的输出PCollection直到处理完成。
输出最终经过transform的PCollction(s)，典型的做法就是使用sink将数据写入外部存储。
运行pipeline使用设计好的Pipeline Runner。
当您运行Beam river程序时，您指定的Pipeline Runner将基于您创建的PCollection对象和您应用的transform构建pipeline的工作流图形。 然后，使用适当的分布式处理后端执行该图，从而在该后端上成为异步“作业”（或等效的）。
2、创建管道
Pipeline抽象封装了数据处理任务中的所有数据和步骤。 您的Beam driver程序通常首先构建一个Pipeline对象，然后使用该对象作为将流水线的数据集创建为PCollections及其操作为Transforms的基础。
为了使用Beam，driver 程序首先需要创建Beam SDK类Pipeline的实例。当创建Pipeline的时候，也需要设置一些配型选项。您可以以编程方式设置Pipeline的配置选项，但通常更容易提前设置选项（或从命令行读取），并在创建对象时将它们传递给Pipeline对象。
Pipeline的配置选项决定了PipelineRunner，并且确定在哪里执行，本地还是选择的分布式后端。根据您的pipeline执行的位置和指定的Runner需要什么，这些选项还可以帮助您指定执行的其他方面。
为了设置你的pipeline配置选项以及创建pipeline，创建一个PipelineOptions类型的对象并且传递给Pipeline.Create()。最常见的就是从命令行解析参数：
public static void main(String[] args) {
// Will parse the arguments passed into the application and construct a PipelineOptions
// Note that --help will print registered options, and --help=PipelineOptionsClassName
// will print out usage for the specific class.
PipelineOptions options =
PipelineOptionsFactory.fromArgs(args).create();
Pipeline p = Pipeline.create(options);
Beam SDK包含与不同Runners对应的PipelineOptions的各种子类。例如，DirectPipelineOptions包含Direct pipeline runner的选项，DataflowPipelineOptions包含使用Google Cloud DataFlow runner的选项。也可以定义自己的PipelineOptions通过继承Bean SDK的PipelineOptions类。
3、使用PCollection
PCollection抽象代表着一个潜在的分布式，多元素数据集。可以把一个PCollection看做“pipeline”的数据；Beam transfor使用PCollection对象作为输入输出。因此，要想在pipeline中的数据工作，必须以PCollection的形式。
在创建完Pipeline，需要以某种格式创建至少一个PCollection。您创建的PCollection将作为管道中第一个操作的输入。
3.1、创建PCollection
您可以通过使用Beam的Source API从外部源读取数据来创建PCollection，也可以在驱动程序中创建存储在内存中集合类中的数据的PCollection。前者通常是生产流水线获取数据的途径; Beam的Source API包含适配器，以帮助您从外部源（如大型基于云的文件，数据库或订阅服务）读取。 后者主要用于测试和调试目的。
3.1.1、从外部数据读取数据
为了能从外部数据读取数据，要使用Beam-provided I/O中提供的一个适配器。适配器的确切用法各不相同，但所有这些适配器都来自某些外部数据源，并返回一个PCollection，其元素表示该Source中的数据记录。
每个数据源适配器具有Read transform; 要读取，您必须将该transform应用于Pipeline对象本身。 例如，TextIO.Read从外部文本文件读取并返回其元素类型为String的PCollection，每个String表示文本文件中的一行。 这里是如何应用TextIO.Read到您的管道创建PCollection：
public static void main(String[] args) {
// Create the pipeline.
PipelineOptions options =
PipelineOptionsFactory.fromArgs(args).create();
Pipeline p = Pipeline.create(options);
PCollection<String> lines = p.apply(
"ReadMyFile", TextIO.Read.from("protocol://path/to/some/inputData.txt"));
}
3.1.2、从内存数据创建PCollection
从内存的JAVA集合创建PCollection，使用Bea提供的Create transform。就像数据适配器的读操作，你可以应用Create操作直接在你的Pipeline对象本身。
作为参数，Create接收java集合以及Coder对象。Coder表明了集合中元素的编码。
以下代码显示如何从内存List创建PCollection：
public static void main(String[] args) {
// Create a Java Collection, in this case a List of Strings.
static final List<String> LINES = Arrays.asList(
"To be, or not to be: that is the question: ",
"Whether 'tis nobler in the mind to suffer ",
"The slings and arrows of outrageous fortune, ",
"Or to take arms against a sea of troubles, ");
// Create the pipeline.
PipelineOptions options =
PipelineOptionsFactory.fromArgs(args).create();
Pipeline p = Pipeline.create(options);
// Apply Create, passing the list and the coder, to create the PCollection.
p.apply(Create.of(LINES)).setCoder(StringUtf8Coder.of())
}
3.2、PCollection的特征
PCollection是由创建它的那个pipline对象所拥有的；多个Pipelines不能共享一个PCollection。在某些方面,PCollection功能想一个集合类。然而，在如下几个方面是不同于集合类的：
元素类型
PCollection的元素可能是任何类型，但还是所有的元素必须是同一种类型，然而，为了支持分布式处理,Beam需要可以把每个单独的元素编码成一个byte字符串（以便于元素可以在分布式workers中传递）。Beam SDK提供了一种数据编码机制，包括常用类型的内置编码以及支持根据需要指定自定义编码。
不变性
PCollection是不可变的，一旦创建就不能增加，移除或者修改元素。Bean transform可以处理PCollection的每个元素并且生成行的pipeline数据（作为行的PCollection），但是不会消费挥着修改原始的输入数据。
随机读写
PCollection不支持单个元素的随机读写。同时，Beam Transform 会单独开率PCollection中的每个元素。
大小与有界性
PCollection是一个大的，不可修改的元素“袋子”，一个PCollection可以容纳的元素个数是没有上限的；任何给定的PCollection可能适合在单个机器上的内存中，或者它可能表示由持久数据存储支持的非常大的分布式数据集。
PCollection在大小上可以是有边界的或者是没有边界的。有边界的PCollection表示一个已知的大小固定的数据集。没有界限的PCollection代表着没有大小限制的数据集。一个PCollection是否是有界或者无界依赖于它所代表的数据集的源。从批处理数据源（例如文件或数据库）读取时，会创建有界PCollection。 从流式或连续更新的数据源（例如Pub / Sub或Kafka）读取将创建一个无界PCollection（除非你明确告诉它不）。
您的PCollection的有界（或无界）性质影响Beam如何处理您的数据。 可以使用批处理作业来处理有界PCollection，其可以读取整个数据集一次，并且在有限长度的作业中执行处理。 必须使用连续运行的流式作业处理无界PCollection，因为整个收集永远不能在任何时间进行处理。
当执行将元素组合在无界PCollection中的操作时，Beam需要一个称为窗口化的概念来将连续更新的数据集划分为有限大小的逻辑窗口。 Beam将每个窗口作为一个包来处理，并且在生成数据集时继续处理。 这些逻辑窗口由与数据元素相关联的某些特性（例如时间戳）确定。
元素时间戳
PCollection中的每个元素都有一个相关的固有时间戳。 每个元素的时间戳最初由创建PCollection的源分配。 创建无限PCollection的源通常为每个新元素分配与元素被读取或添加时相对应的时间戳。
注意：为固定数据集创建有界PCollection的源也会自动分配时间戳，但最常见的行为是为每个元素分配相同的时间戳（Long.MIN_VALUE）。
时间戳对于包含具有固有时间概念的元素的PCollection是有用的。 如果您的管道正在读取事件流（如Tweets或其他社交媒体消息），则每个元素可能会将事件发布的时间用作元素时间戳。
如果源不为提供时间处理，您可以手动将时间戳分配给PCollection的元素。 如果元素具有固有的时间戳，但时间戳在元素本身的结构中的某处（例如服务器日志条目中的“时间”字段），您将需要执行此操作。 Beam的变换会采用PCollection作为输入并输出具有附加的时间戳的相同PCollection; 有关如何执行此操作的详细信息，请参阅分配时间戳。
4、应用变换
在Beam SDK中，transform是pipeline中的操作。transform会采用一个后者多个PCollection作为输入，在collection上的每个元素执行指定的操作，并且产生一个新的输出PCollection。为了调用一个变换，必须将他应用在PCollection上。
在Beam SDK中每个transform都有一个apply方法。调用多个Beam transforms和方法链（method chaining）类似，但是有一点不同：将transform应用于输入PCollection，将transform本身作为参数传递，并且操作返回输出PCollection。 这采取一般形式：
[Output PCollection] = [Input PCollection].apply([Transform])
由于Beam使用通用的apply作用于PCollection，您可以按顺序链接transform，也可以应用包含嵌套在其中的其他transform的transform（在Beam SDK中称为复合transform）。
你怎么应用你的pipeline的transform决定着你pipeline的结构。考虑你的pipline的最好方式是一个有向无环图，其中节点是PCollections，边缘是transform。 例如，您可以chain transform以创建顺序管道，如下所示：
[Final Output PCollection] = [Initial Input PCollection].apply([First Transform])
.apply([Second Transform])
.apply([Third Transform])
上述管道的结果工作流图如下所示：
[Sequential Graph Graphic]
但是，请注意，transform不会消耗或以其他方式更改输入集合 - 记住，PCollection根据定义是不可变的。 这意味着您可以将多个transform应用于同一个输入PCollection以创建分支管道，如下所示
[Output PCollection 1] = [Input PCollection].apply([Transform 1])
[Output PCollection 2] = [Input PCollection].apply([Transform 2])
上述分支pipeline的结果工作流图如下所示：
[Branching Graph Graphic]
您还可以构建自己的复合变换，在单个较大的变换中嵌套多个子步骤。 复合变换对于构建可重用的简单步骤序列特别有用，可以在许多不同的地方使用。
4.1、BeamSDK中的transform
在BeamSDK中的transform提供了一个通用的处理框架，你只需要提供以函数对象形式提供的处理逻辑。用户代码会作用于输入PCollection的每个元素。用户代码的实例可能会在集群的不同工作节点上并行执行。这些依赖于执行Beam pipeline的pipeline runner以及你选择的后端程序。在没和工作节点的用户代码产生的输出元素并且最终会输出到tansform产生的PCollection中。
4.2、Beam核心的transform
Beam提供了一下的transform,每个都代表着不同的处理案例：
ParDo
GroupByKey
Combine
Flatten and Partition
ParDo
Pado是一个通用的并行beam transform处理操作。ParDo处理和“Map/Shuffle/Reduce-style”算法的“Map”类似：一个ParDo transform会作用于输入PCollection的每个元素，在元素上执行一些指定的函数并且输出0个，1个挥着多个元素到输出PCollection。
ParDo可用于各种常见的数据处理操作，包括：
#、过滤数据集.你可以使用ParDo来考虑在PCollection中的每个元素并且决定是否输出到PCollection中或者忽略它。
#、对数据集中的每个元素格式化或者类型转换。如果输入PCOllection中的元素和你想要的元素是不同的类型或者格式；可以使用ParDo在每个元素上执行转换输出结果到PCollection中。
#、抽取数据集中的一部分数据。例如，如果有一个PCollection，它的记录有多个字段，可以使用ParDo解析出自己想要的字段来输出到PCollection中。
#、在数据集中的每个元素执行计算。可以使用ParDo来在PCollection每个元素上或者特定的元素上执行简单或者复杂的计算，输出一个新的PCollection。
在这些情况下，ParDo在pipeline中是一个普通的中间步骤。你也可以使用它来从输入的原始数据中提取指定的字段或者转换原始数据到另一种形式；你也可以使用ParDo来转换处理数据到另一种合适的形式输出，就像表行后者可打印的字符串。
当使用ParDo transform的时候，需要提供给以DoFn对象性的用户代码。DoFn是一个Beam SDK类，定义了一个分布式处理函数。
注意：当你创建DoFn的子类，你的子类需要遵守https://beam.apache.org/documentation/programming-guide/#transforms-usercodereqs
ParDo的应用
就像其他Beam Transform一样，可以通过调用apply方法来使用ParDo在输出PCollection上并且传递ParDo来作为一个参数，如下示例代码所示：
// The input PCollection of Strings.
PCollection<String> words = ...;
// The DoFn to perform on each element in the input PCollection.
static class ComputeWordLengthFn extends DoFn<String, Integer> { ... }
// Apply a ParDo to the PCollection "words" to compute lengths for each word.
PCollection<Integer> wordLengths = words.apply(
ParDo
.of(new ComputeWordLengthFn())); // The DoFn to perform on each element, which
// we define above.
在这个例子中，我们输出PCollection包含了String 值，使用ParDo transform来指定一个函数计算每一个字符串的长度，输出结果到新的整数值得PCollection中。
创建DoFn
传递给ParDo的DoFn对象包含了应用在输入集合上的处理逻辑。当使用Beam的时候，最重要的就是你要写的代码片段—它们会定义pipeline的准确的数据处理任务。
注意：当你创建DoFn的时候，你的需要遵守https://beam.apache.org/documentation/programming-guide/#transforms-usercodereqs
DoFn一次处理输入PCollection中的一个元素。当创建DoFn的一个子类的售后，需要输入类型参数来平匹配输入输出数据类型。如果DoFn处理进来的String类型的元素并且输出会产生Integer元素，你的类型声明看起来像这样：
static class ComputeWordLengthFn extends DoFn<String, Integer> { ... }
在DoFn子类内部，需要定义一个带有注解@ProcessElement的方法，这个方法提供了实际的处理逻辑。你不必手工在输入集合中抽取元素；Beam SDKs已经为你处理完了。ProcessContext对象允许您访问输入元素并且提供输出元素的方法：
static class ComputeWordLengthFn extends DoFn<String, Integer> {
@ProcessElement
public void processElement(ProcessContext c) {
// Get the input element from ProcessContext.
String word = c.element();
// Use ProcessContext.output to emit the output element.
c.output(word.length());
}
}
Note: 如果在输入PCollotion中的元素是key/value对，可以使用ProcessContext.element().getKey()或者ProcessContext.element().getValue()来分别获取key或者value。
给定的DoFn实例通常被调用一次或多次以处理一些任意的元素束。 但是，Beam不保证调用的确切数量; 它可以在给定工作节点上多次调用以解决故障和重试。 因此，您可以在处理方法的多次调用的时缓存信息，但如果这样做，请确保实现不依赖于调用数。
在处理方法中，您还需要满足一些不可变性要求，以确保Beam和处理后端可以安全地序列化和缓存管道中的值。 您的方法应该满足以下要求：
#、 您不应以任何方式修改ProcessContext.element（）或ProcessContext.sideInput（）（来自输入集合的传入元素）返回的元素。
#、 一旦使用ProcessContext.output（）或ProcessContext.sideOutput（）输出值，则不应以任何方式修改该值。
轻量的DoFns以及其他的抽象
如果你的函数相对简单，你可以通过提供一个轻量级的DoFn in-line作为一个匿名内部类实例来简化ParDo的使用。
这里是前面的例子，ParDo with ComputeLengthWordsFn，将DoFn指定为匿名内部类实例：
// The input PCollection.
PCollection<String> words = ...;
// Apply a ParDo with an anonymous DoFn to the PCollection words.
// Save the result as the PCollection wordLengths.
PCollection<Integer> wordLengths = words.apply(
"ComputeWordLengths", // the transform name
ParDo.of(new DoFn<String, Integer>() { // a DoFn as an anonymous inner class instance
@ProcessElement
public void processElement(ProcessContext c) {
c.output(c.element().length());
}
}));
如果ParDo执行输入元素到输出元素的一对一映射，即对于每个输入元素，它应用一个只产生一个输出元素的函数，则可以使用更高级别的MapElements变换。 MapElements可以接受一个匿名Java 8 lambda函数，以便更加简洁。
以下是使用MapElements的上一个示例：
// The input PCollection.
PCollection<String> words = ...;
// Apply a MapElements with an anonymous lambda function to the PCollection words.
// Save the result as the PCollection wordLengths.
PCollection<Integer> wordLengths = words.apply(
MapElements.via((String word) -> word.length())
.withOutputType(new TypeDescriptor<Integer>() {});
NOTE：Filter,FlatMapElements和Partition这些Beam transform都可以实用Java lamada表达式。
使用GroupByKey
GroupByKey是用于处理键/值对的集合的Beam transform。这是一个并行缩减操作，类似于Map / Shuffle / Reduce风格算法的Shuffle阶段。 GroupByKey的输入是表示多重映射的键/值对的集合，其中集合包含具有相同键但不同值的多个对。给定这样的集合，您使用GroupByKey来收集与每个唯一键相关联的所有值。
GroupByKey是一种聚合具有共同点的数据的好方法。例如，如果您有一个存储客户订单记录的集合，您可能希望将来自相同邮政编码的所有订单分组在一起（其中键/值对的“键”是邮政编码字段，而“值“是记录的剩余部分）。
让我们用一个简单的例子来研究GroupByKey的机制，其中我们的数据集包括文本文件中的单词和它们出现的行号。我们希望将所有共享同一个单词（键）的行号（值）组合在一起，让我们看到文本中出现特定单词的所有地方。
我们的输入是一个键/值对的PCollection，其中每个单词是一个键，值是文件中出现该单词的行号。以下是输入集合中的键/值对的列表：
cat, 1
dog, 5
and, 1
jump, 3
tree, 2
cat, 5
dog, 2
and, 2
cat, 9
and, 6
...
GroupByKey使用相同的键收集所有值，并输出一个新对，该对由唯一键和与输入集合中的该键相关联的所有值的集合组成。 如果我们将GroupByKey应用到上面的输入集合中，输出集合将如下所示：
cat, [1,5,9]
dog, [5,2]
and, [1,2,6]
jump, [3]
tree, [2]
...
因此，GroupByKey代表一个从multimap(多个键到单个值 )到一个uni-map(唯一键对应值得集合)的transform。
注意：Beam表示键/值对略有不同，具体取决于您使用的语言和SDK。 在用于Java的Beam SDK中，您表示具有类型KV <K，V>的对象的键/值对。 在Python中，你用2元组表示键/值对。
使用Combine
Combine是用于组合数据集合中的元素或值的Beam transform。Combine具有在整个PCollection上工作的变体，并且会组合PCollections中键/值对的每个键的值。
应用Combine transform时，必须提供包含用于组合元素或值的逻辑的函数。组合函数应该是可交换的和关联的，因为该函数不一定对具有给定关键字的所有值调用一次。因为输入数据（包括值集合）可以分布在多个工作节点上，所以可以多次调用组合函数以对值集合的子集执行部分组合。 Beam SDK还为常见的数字组合操作（如sum，min和max）提供了一些预构建的组合函数。
简单的合并操作，如sum，通常可以实现为一个简单的函数。更复杂的组合操作可能需要您创建一个具有与输入/输出类型不同的累积类型的CombineFn子类。
1）、使用简单函数的简单合并
如下代码显示了简单的合并函数：
// Sum a collection of Integer values. The function SumInts implements the interface SerializableFunction.
public static class SumInts implements SerializableFunction<Iterable<Integer>, Integer> {
@Override
public Integer apply(Iterable<Integer> input) {
int sum = 0;
for (int item : input) {
sum += item;
}
return sum;
}
}
2）、使用CombineFn的高级合并
对于更复杂的组合函数，您可以定义CombineFn的子类。 如果组合函数需要更复杂的累加器，则必须使用CombineFn，必须执行额外的预处理或后处理，可能会更改输出类型或考虑key。
一般组合操作包括四个操作。 当您创建CombineFn的子类时，必须通过覆盖相应的方法提供四个操作：
#、创建累加器:创建一个新的“本地”累加器。 在示例情况下，采用平均值，本地累加器跟踪值的运行总和（用于我们的最终平均除法的分子值）和到目前为止总计的值的数量（分母值）。 它可以以分布式方式被调用任意次数。
#、添加输入；添加一个输入元素到一个累加其中，返回累加值。在我们的例子中，它要更新和以及增加计数值。这个也可能并行执行。
#、合并累加器：合并多个累加器为一个累加器；这就是在最终计算之前在多个累加器中的数据怎么合并的。在平平均计算的情况下，表示被分割的每个部分的累加器被合并在一起。 可以在其输出上再次调用任何次数。
#、提取输出：执行最终的计算。在计算平均值的情况下，这意味着将所有值的组合和除以所求和的值的数量。 它在最终的合并累加器上调用一次。
如下代码显示了怎样定义一个CombinFn行数据计算平均值：
public class AverageFn extends CombineFn<Integer, AverageFn.Accum, Double> {
public static class Accum {
int sum = 0;
int count = 0;
}
@Override
public Accum createAccumulator() { return new Accum(); }
@Override
public Accum addInput(Accum accum, Integer input) {
accum.sum += input;
accum.count++;
return accum;
}
@Override
public Accum mergeAccumulators(Iterable<Accum> accums) {
Accum merged = createAccumulator();
for (Accum accum : accums) {
merged.sum += accum.sum;
merged.count += accum.count;
}
return merged;
}
@Override
public Double extractOutput(Accum accum) {
return ((double) accum.sum) / accum.count;
}
}
如果您组合key-value形式的键值对的PCollection，每键组合通常就足够了。 如果您需要根据key更改合并策略（例如，某些用户为MIN，其他用户为MAX），则可以定义一个KeyedCombineFn以访问合并策略中的键。
3）、组合PCollection到一个单独的值
使用全局组合将给定PCollection中的所有元素转换为单个值，在您的pipeline中表示为包含一个元素的新PCollection。 以下示例代码显示了如何应用“Bean提供的sum”组合函数，为整数的PCollection生成单个和值。
// Sum.SumIntegerFn() combines the elements in the input PCollection.
// The resulting PCollection, called sum, contains one value: the sum of all the elements in the input PCollection.
PCollection<Integer> pc = ...;
PCollection<Integer> sum = pc.apply(
Combine.globally(new Sum.SumIntegerFn()));
4)、全局窗口
如果你的输入PCollection使用的默认的全局窗口，默认的行为是返回一个包含一个条目的PCollection。该条目的值来自在应用Combine时指定的combine函数中的累加器。 例如，Beam提供的sum组合函数返回零值（空输入的和），而min组合函数返回最大或无限值。
如果输入为空，要使Combine返回一个空PCollection，您应用Combine transform，指定.withoutDefaults时，如下面的代码示例：
PCollection<Integer> pc = ...;
PCollection<Integer> sum = pc.apply(
Combine.globally(new Sum.SumIntegerFn()).withoutDefaults());
5）、非全局窗口
如果你PCollection使用非全局窗口函数，Beam不会提供默认的行为。当使用Combine的时候你必须指定以下选项中的一个：
#、指定.withoutDefaults，其中输入PCollection中为空的窗口在输出集合中将同样为空。
#、指定.asSingletonView，其中输出立即转换为PCollectionView，当用作侧输入（side input）时，它将为每个空窗口提供默认值。 如果管道的Combine的结果要在管道中稍后用作边输入(side input)，则通常只需要使用此选项。
6)、合并key分组集合的值
在创建键分组集合（例如，通过使用GroupByKey变换）之后，公共模式是将与每个键相关联的值的集合组合成单个合并的值。 基于GroupByKey的上一个示例，一个键组分组的PCollection称为groupedWords如下所示：
cat, [1,5,9]
dog, [5,2]
and, [1,2,6]
jump, [3]
tree, [2]
...
在上面的PCollection中，每个元素都有一个字符串键（例如“cat”）和一个可迭代的整数值（在第一个元素中，包含[1,5,9]）。 如果我们的流水线的下一个处理步骤组合这些值（而不是单独考虑它们），您可以组合整数的可迭代，以创建一个单一的合并值与每个键配对。 GroupByKey的这种模式，然后合并值的集合等效于Beam的Combine PerKey变换。 您提供给Combine PerKey的组合函数必须是关联缩减函数或CombineFn的子类。
// PCollection is grouped by key and the Double values associated with each key are combined into a Double.
PCollection<KV<String, Double>> salesRecords = ...;
PCollection<KV<String, Double>> totalSalesPerPerson =
salesRecords.apply(Combine.<String, Double, Double>perKey(
new Sum.SumDoubleFn()));
// The combined value is of a different type than the original collection of values per key.
// PCollection has keys of type String and values of type Integer, and the combined value is a Double.
PCollection<KV<String, Integer>> playerAccuracy = ...;
PCollection<KV<String, Double>> avgAccuracyPerPlayer =
playerAccuracy.apply(Combine.<String, Integer, Double>perKey(
new MeanInts())));
使用Flatten以及Partition
Flatten 和 Partition 是存储相同数据类型的PCollection对象的Beam transform。 Flatten将多个PCollection对象合并为单个逻辑PCollection，并且Partition将单个PCollection分割为固定数量的较小集合。
1）、Flatten
以下代码表明了如何使用Flatten transform来合并多个PCollection对象：
// Flatten takes a PCollectionList of PCollection objects of a given type.
// Returns a single PCollection that contains all of the elements in the PCollection objects in that list.
PCollection<String> pc1 = ...;
PCollection<String> pc2 = ...;
PCollection<String> pc3 = ...;
PCollectionList<String> collections = PCollectionList.of(pc1).and(pc2).and(pc3);
PCollection<String> merged = collections.apply(Flatten.<String>pCollections());
在合并集合中的数据编码：
输出PCollection的编码默认的是和输入PCollectionList中的第一个PCollection的编码是相同的。然而，输入PCollection对象可以使用不同的编码器，只要它们都包含您选择的语言中相同的数据类型。
合并窗口集合：
当使用Flatten合并应用了窗口策略的PCollection对象时，要合并的所有PCollection对象必须使用兼容的窗口策略和窗口大小。 例如，您合并的所有集合必须使用（假设）相同的5分钟固定窗口或每30秒开始的4分钟滑动窗口。
如果您的管道尝试使用Flatten将PCollection对象与不兼容的窗口合并，则Beam在构建管道时会生成IllegalStateException错误。
2）、Partition
分区根据您提供的分区函数划分PCollection的元素。 分区函数包含确定如何将输入PCollection的元素拆分成每个结果分区PCollection的逻辑。 分区的数量必须在图构造时确定。 例如，您可以在运行时将分区数作为命令行选项传递（这将用于构建管道图），但是您无法确定中间管道中的分区数（例如，基于之后计算的数据 你的管道图被构造）。
以下示例将PCollection分为百分位数组。
// Provide an int value with the desired number of result partitions, and a PartitionFn that represents the partitioning function.
// In this example, we define the PartitionFn in-line.
// Returns a PCollectionList containing each of the resulting partitions as individual PCollection objects.
PCollection<Student> students = ...;
// Split students up into 10 partitions, by percentile:
PCollectionList<Student> studentsByPercentile =
students.apply(Partition.of(10, new PartitionFn<Student>() {
public int partitionFor(Student student, int numPartitions) {
return student.getPercentile() // 0..99
* numPartitions / 100;
}}));
// You can extract each partition from the PCollectionList using the get method, as follows:
PCollection<Student> fortiethPercentile = studentsByPercentile.get(4);
一般要求为写入Beam变换的用户代码
当你构建一个Beam transform的用户代码时，你应该记住执行的分布式性质。 例如，可能有许多您的功能的副本在许多不同的机器上并行运行，并且这些副本独立运行，而不与任何其他副本通信或共享状态。 根据pipeline运行程序和为pipeline选择的后端处理，可以重试或运行用户代码函数的每个副本多次。 因此，您应该谨慎地在您的用户代码中包含诸如状态依赖性之类的内容。
从大体上来讲，用户代码至少实现含如下的要求：
#、函数对象必须可以实例化。
#、函数对象必须是线程兼容的，以及意识到Beam SDK不是线程安全的。
此外，它建议你使你的函数对象幂等。
注意：这些要求适用于DoFn（与ParDo变换一起使用的函数对象）的子类，CombineFn（与Combine变换一起使用的函数对象）和WindowFn（与Window变换一起使用的函数对象）。
1）、序列化
您为转换提供的任何函数对象必须是完全可序列化的。 这是因为函数的副本需要序列化并传输到处理集群中的远程工作线程。 用户代码的基类（如DoFn，CombineFn和WindowFn）已实现Serializable; 但是，您的子类不能添加任何非序列化成员。
一些其他的序列化因子应该记住的：
#、函数对象中的transient字段不会传输到工作线程实例，因为它们不会自动序列化。
#、避免在序列化之前加载大量数据的字段。
#、您的函数对象的各个实例不能共享数据。
#、在应用函数对象之后对其进行transform将没有任何效果。
#、在使用匿名内部类实例声明您的函数对象时，请小心。 在非静态上下文中，内部类实例将隐式包含一个指向封闭类和该类的状态的指针。 该封装类也将被序列化，因此应用于函数对象本身的相同注意事项也适用于此外部类。
2）、线程兼容性
你的函数对象应该是线程兼容的。 您的函数对象的每个实例都由worker实例上的单个线程访问，除非您显式创建自己的线程。 但请注意，Beam SDK不是线程安全的。 如果在用户代码中创建自己的线程，则必须提供自己的同步。 请注意，函数对象中的静态成员不会传递给worker实例，并且可以从不同的线程访问函数的多个实例。
3）、幂等
建议您使您的功能对象幂等，即，它可以重复或重试，如有必要，而不会导致意想不到的副作用。 Beam模型不能保证您的用户代码可能被调用或重试的次数; 因此，保持您的函数对象幂等性保持您的管道的输出确定性，并且您的变换的行为更可预测和更容易调试。
side inputs and side uptputs
1）、side inputs
除了主输入PCollection之外，您还可以以side inputs的形式为ParDo变换提供附加输入。 side inputs是您的DoFn每次处理输入PCollection中的元素时可以访问的附加输入。 当指定side inputs时，您将创建一个其他数据的视图，在处理每个元素时，可以从ParDo变换的DoFn中读取。
如果您的ParDo需要在处理输入PCollection中的每个元素时插入附加数据，但是附加数据需要在运行时确定（而不是硬编码），则side inputs非常有用。 这些值可以由输入数据确定，或者取决于管道的不同分支。
传递side inputs给ParDo：
// Pass side inputs to your ParDo transform by invoking .withSideInputs.
// Inside your DoFn, access the side input by using the method DoFn.ProcessContext.sideInput.
// The input PCollection to ParDo.
PCollection<String> words = ...;
// A PCollection of word lengths that we'll combine into a single value.
PCollection<Integer> wordLengths = ...; // Singleton PCollection
// Create a singleton PCollectionView from wordLengths using Combine.globally and View.asSingleton.
final PCollectionView<Integer> maxWordLengthCutOffView =
wordLengths.apply(Combine.globally(new Max.MaxIntFn()).asSingletonView());
// Apply a ParDo that takes maxWordLengthCutOffView as a side input.
PCollection<String> wordsBelowCutOff =
words.apply(ParDo.withSideInputs(maxWordLengthCutOffView)
.of(new DoFn<String, String>() {
public void processElement(ProcessContext c) {
String word = c.element();
// In our DoFn, access the side input.
int lengthCutOff = c.sideInput(maxWordLengthCutOffView);
if (word.length() <= lengthCutOff) {
c.output(word);
}
}}));
Side input 与 窗口：
一个窗口PCollection可能是无穷的，因此不能够压缩成一个单独的值。当创建窗口PCollection的PCollectionView时，PCollectionView表示每个窗口的单个实体（每个窗口一个单例，每个窗口一个列表等）。
Beam使用主输入元素的窗口来查找side input元素的相应窗口。Beam将主输入元素的窗口投影到side input的窗口集中，然后使用来自结果窗口的side input。如果主输入和side input具有相同的窗口，则投影提供精确对应的窗口。但是，如果输入具有不同的窗口，则Beam使用投影选择最合适的side input窗口。
例如，如果主输入使用一分钟的固定时间窗口窗口化，而side input使用一小时的固定时间窗口窗口化，则Beam将主输入窗口相对于side input窗口集合投影，并选择side input值从适当的小时长边输入窗口。
如果主输入元素存在于多个窗口中，则processElement将被调用多次，每次调用一次。每次调用processElement都会为主输入元素投影“当前”窗口，因此可能会每次提供不同的side input视图。
如果side input具有多个触发器触发，则Beam使用来自最近触发器触发的值。如果使用具有单个全局窗口的side input并指定触发器，这将特别有用。
2）、side output
ParDo总是产生一个主输出PCollection，然而也可以使ParDo产生任意数量的附加输出PCollections。如果想有多个输出，ParDo会返回所有绑定在一起的PCollection（包括主输入）。
side output 的 Tags：
// To emit elements to a side output PCollection, create a TupleTag object to identify each collection that your ParDo produces.
// For example, if your ParDo produces three output PCollections (the main output and two side outputs), you must create three TupleTags.
// The following example code shows how to create TupleTags for a ParDo with a main output and two side outputs:
// Input PCollection to our ParDo.
PCollection<String> words = ...;
// The ParDo will filter words whose length is below a cutoff and add them to
// the main ouput PCollection<String>.
// If a word is above the cutoff, the ParDo will add the word length to a side output
// PCollection<Integer>.
// If a word starts with the string "MARKER", the ParDo will add that word to a different
// side output PCollection<String>.
final int wordLengthCutOff = 10;
// Create the TupleTags for the main and side outputs.
// Main output.
final TupleTag<String> wordsBelowCutOffTag =
new TupleTag<String>(){};
// Word lengths side output.
final TupleTag<Integer> wordLengthsAboveCutOffTag =
new TupleTag<Integer>(){};
// "MARKER" words side output.
final TupleTag<String> markedWordsTag =
new TupleTag<String>(){};
// Passing Output Tags to ParDo:
// After you specify the TupleTags for each of your ParDo outputs, pass the tags to your ParDo by invoking .withOutputTags.
// You pass the tag for the main output first, and then the tags for any side outputs in a TupleTagList.
// Building on our previous example, we pass the three TupleTags (one for the main output and two for the side outputs) to our ParDo.
// Note that all of the outputs (including the main output PCollection) are bundled into the returned PCollectionTuple.
PCollectionTuple results =
words.apply(
ParDo
// Specify the tag for the main output, wordsBelowCutoffTag.
.withOutputTags(wordsBelowCutOffTag,
// Specify the tags for the two side outputs as a TupleTagList.
TupleTagList.of(wordLengthsAboveCutOffTag)
.and(markedWordsTag))
.of(new DoFn<String, String>() {
// DoFn continues here.
...
}
在DoFn中发射到侧面输出：
// Inside your ParDo's DoFn, you can emit an element to a side output by using the method ProcessContext.sideOutput.
// Pass the appropriate TupleTag for the target side output collection when you call ProcessContext.sideOutput.
// After your ParDo, extract the resulting main and side output PCollections from the returned PCollectionTuple.
// Based on the previous example, this shows the DoFn emitting to the main and side outputs.
.of(new DoFn<String, String>() {
public void processElement(ProcessContext c) {
String word = c.element();
if (word.length() <= wordLengthCutOff) {
// Emit this short word to the main output.
c.output(word);
} else {
// Emit this long word's length to a side output.
c.sideOutput(wordLengthsAboveCutOffTag, word.length());
}
if (word.startsWith("MARKER")) {
// Emit this word to a different side output.
c.sideOutput(markedWordsTag, word);
}
}}));
5、复合变换 1）、Pipeline I/O
当创建pipeline，你学从外部数据源读取数据，例如从文件或者数据库。同样的，你也可能想你的pipeline输出数据到类似的外部数据sink。Beam为许多常见的数据存储类型提供读写transform。 如果希望pipeline读取或写入内置transform不支持的数据存储格式，可以实现自己的读写transform。
2）、读取输入数据
读取transform从外部源读取数据，并返回数据的PCollection表示形式供pipeline使用。 您可以在构建pipeline时随时使用读取transform来创建新的PCollection，尽管它在pipeline开始时最为常见。
使用读取transform：
PCollection<String> lines = p.apply(TextIO.Read.from("gs://some/inputData.txt"));
3)、写输出数据
写transform将PCollection中的数据写入外部数据源。 你最常使用在pipeline结尾处的写transform来输出pipeline的最终结果。 但是，您可以使用写transform在pipeline中的任何点输出PCollection的数据。
output.apply(TextIO.Write.to("gs://some/outputData"));
4）、基于文件的输入输出数据
从多个本地文件读取数据：
许多读取transform支持从匹配您提供的glob运算符的多个输入文件读取。 请注意，glob运算符是特定于文件系统的并且遵循文件系统特定的一致性模型。 以下TextIO示例使用glob运算符（*）来读取在给定位置具有前缀“input-”和后缀“.csv”的所有匹配的输入文件：
p.apply(“ReadFromText”,
TextIO.Read.from("protocol://my_bucket/path/to/input-*.csv");
要将来自不同来源的数据读入单个PCollection，请独立读取每个PCollection，然后使用Flatten变换创建单个PCollection。
写入多个输出文件：
对于基于文件的输出数据，默认情况下，写入转换写入多个输出文件。 将输出文件名传递到写入转换时，文件名将用作写入转换生成的所有输出文件的前缀。 您可以通过指定后缀为每个输出文件附加后缀。
以下写入转换示例将多个输出文件写入位置。 每个文件都有前缀“numbers”，数字标签和后缀“.csv”。
5)、Beam提供的I/O APIs
请参阅Beam支持的I / O API的特定于语言的源代码目录。 将来会添加这些I / O源中的每个的具体文档。
Language File-based Messaging Database
Java
AvroIO
HDFS
TextIO
XML
JMS
Kafka
Kinesis
Google Cloud PubSub
Apache HBase
MongoDB
JDBC
Google BigQuery
Google Cloud Bigtable
Google Cloud Datastore
Python
avroio
textio
Google BigQuery
Google Cloud Datastore
6）、运行pipeline
使用run方法来运行pipeline。您创建的程序将您的pipeline的规范发送给pipeline运行程序，然后pipeline运行程序构建和运行实际的pipeline操作系列。 默认情况下，pipeline是异步执行的。
pipeline.run();
pipeline.run().waitUntilFinish();
6、数据编码与类型安全
当你创建或者输出pipeline数据的时候，你需要指定在PCollection中元素的编码以及从byte 字符串中解码。Byte字符串用于中间存储以及从source和写入到sink。Beam 使用coders对象来描述一个给定的PCollection进行编码与解码。
1）、使用coders
通常需要在从外部源读取数据到pipeline（或从本地数据创建pipeline数据）时指定编码器，以及在将pipeline数据输出到外部sink时。
在用于Java的Beam SDK中，类型Coder提供了编码和解码数据所需的方法。 Java的SDK提供了许多Coder子类，它们可以与各种标准Java类型（如Integer，Long，Double，StringUtf8等）配合使用。您可以在编码器包中找到所有可用的编码器子类。
当您将数据读入pipline时，编码器指示如何将输入数据解释为特定于语言的类型，例如整数或字符串。同样，编码器指示如何将pipeline中的语言特定类型写入输出数据sink的字节字符串中，或者实现pipeline中的中间数据。
Beam SDK为pipeline中的每个PCollection设置一个编码器，包括作为transform的输出生成的编码器。大多数时候，Beam SDK可以自动推断输出PCollection的正确编码器。
注意，编码器不一定与类型有1：1的关系。 例如，Integer类型可以有多个有效的编码器，输入和输出数据可以使用不同的Integer编码器。 transform可能具有使用BigEndianIntegerCoder的整数类型输入数据，以及使用VarIntCoder的整数类型输出数据。
您可以在输入或输出PCollection时明确设置编码器。 在应用pipeline的读或写transform时，通过调用方法.withCoder设置编码器。
通常，当PCollection的编码器无法自动推断时，或者您想使用与管道默认值不同的编码器时，您可以设置编码器。 以下示例代码从文本文件中读取一组数字，并为所得PCollection设置类型为TextualIntegerCoder的编码器：
PCollection<Integer> numbers =
p.begin()
.apply(TextIO.Read.named("ReadNumbers")
.from("gs://my_bucket/path/to/numbers-*.txt")
.withCoder(TextualIntegerCoder.of()));```
您可以使用方法PCollection.setCoder为现有PCollection设置编码器。 请注意，您不能在已经完成的PCollection上调用setCoder（例如，通过调用.apply）。
您可以使用getCoder方法获取现有PCollection的编码器。 如果没有设置编码器，并且不能为给定的PCollection推断，则此方法将失败并返回一个IllegalStateException。
2）、编码推断以及默认编码
Beam SDK需要一个编码器为您的pipeline中的每个PCollection。然而，大多数时候，您不需要显式指定编码器，例如对于由pipeline中间的变换生成的中间PCollection。在这种情况下，Beam SDK可以从用于生成PCollection的变换的输入和输出中推断出适当的编码器。
每个管道对象都有一个CoderRegistry。 CoderRegistry表示Java类型到默认编码器的映射，pipeline应该使用每种类型的PCollections。
默认情况下，Java的Beam SDK自动使用来自变换的函数对象（如DoFn）的type参数推断输出PCollection的元素的编码器。例如，在ParDo的情况下，DoFn <Integer，String>函数对象接受类型为Integer的输入元素，并生成类型为String的输出元素。在这种情况下，Java SDK会自动推断出输出PCollection <String>（默认管道CoderRegistry，这是StringUtf8Coder）的默认编码器。
注意：如果使用创建transform从内存数据创建PCollection，则不能依赖于编码器推理和默认编码器。创建不能访问其参数的任何键入信息，并且如果参数列表包含其精确的运行时类没有注册默认编码器的值，则可能无法推断编码器。
当使用Create时，确保您具有正确编码器的最简单的方法是在应用Create变换时调用withCoder。
3）、默认编码以及CodeRegistry
每个Pipeline对象都有一个CoderRegistry对象，它将语言类型映射到pipeline应该为那些类型使用的默认编码器。您可以自己使用CoderRegistry查找给定类型的默认编码器，或者为给定类型注册新的默认编码器。
CoderRegistry包含对使用Beam SDK for Java创建的任何管道的标准Java类型的编码器默认映射。下表显示标准映射：
Java Type Default Coder
Double DoubleCoder
Instant InstantCoder
Integer VarIntCoder
Iterable IterableCoder
KV KvCoder
List ListCoder
Map MapCoder
Long VarLongCoder
String StringUtf8Coder
TableRow TableRowJsonCoder
Void VoidCoder
byte[ ] ByteArrayCoder
TimestampedValue TimestampedValueCoder
查找默认编码器
您可以使用方法CoderRegistry.getDefaultCoder来确定Java类型的默认编码器。您可以通过使用方法Pipeline.getCoderRegistry访问给定管道的CoderRegistry。这允许您基于每个pipeline确定（或设置）Java类型的默认编码器：即“对于此pipeline，请验证整数值是否使用BigEndianIntegerCoder编码。
设置类型的默认编码器
要为特定pipeline的Java类型设置默认编码器，您需要获取和修改管道的CoderRegistry。您可以使用方法Pipeline.getCoderRegistry获取CoderRegistry对象，然后使用方法CoderRegistry.registerCoder为目标类型注册一个新的编码器。
以下示例代码演示如何设置默认编码器（在本例中为BigEndianIntegerCoder），用于管道的Integer值。
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
CoderRegistry cr = p.getCoderRegistry();
cr.registerCoder(Integer.class, BigEndianIntegerCoder.class);
使用默认编码器注释自定义数据类型
如果您的pipeline程序定义了自定义数据类型，则可以使用@DefaultCoder注释来指定要与该类型一起使用的编码器。 例如，假设您有要使用SerializableCoder的自定义数据类型。 您可以使用@DefaultCoder注释，如下所示：
@DefaultCoder(AvroCoder.class)
public class MyCustomDataType {
...
}
如果你已经创建了自定义的编码器匹配数据类型，并且还想使用@DefaultCoder注解，你的编码器类必须实现一个静态的Coder.of(Class<T>)工厂方法。
public class MyCustomCoder implements Coder {
public static Coder<T> of(Class<T> clazz) {...}
...
}
@DefaultCoder(MyCustomCoder.class)
public class MyCustomDataType {
...
}