Apache Beam -- 编程指南

快来打我* 2022-03-19 03:36 300阅读 0赞

## 指南 ##

Beam Programming Guide适用于希望使用Beam SDK创建数据处理流水线的Beam用户。 它为使用Beam SDK类构建和测试管道提供了指导。 它不是一个详尽的参考，而是作为一种语言无关的高级指南，以编程方式构建您的Beam管道。

## 概述 ##

要使用Beam，您需要首先使用其中一个Beam SDK中的类创建驱动程序。 您的驱动程序定义了您的管道，包括所有输入，转换和输出; 它还为您的管道设置执行选项（通常使用命令行选项传递）。 其中包括Pipeline Runner，它反过来确定您的管道将在哪个后端运行。

Beam SDK提供了许多抽象，简化了大规模分布式数据处理的机制。 相同的Beam抽象适用于批处理和流数据源。 创建Beam管道时，您可以根据这些抽象来考虑数据处理任务。 他们包括：

1.  Pipeline：管道从头到尾封装整个数据处理任务。这包括读取输入数据，转换数据和写入输出数据。所有Beam驱动程序都必须创建管道。创建管道时，还必须指定执行选项，以告知管道在何处以及如何运行。
2.  PCollection：PCollection表示Beam管道操作的分布式数据集。数据集可以是有界的，这意味着它来自固定的源，如文件，或无限制，这意味着它来自通过订阅或其他机制不断更新的源。您的管道通常通过从外部数据源读取数据来创建初始PCollection，但您也可以从驱动程序中的内存数据创建PCollection。从那里，PCollections是管道中每个步骤的输入和输出。
3.  PTransform：PTransform表示管道中的数据处理操作或步骤。每个PTransform都将一个或多个PCollection对象作为输入，执行您在该PCollection的元素上提供的处理函数，并生成零个或多个输出PCollection对象。
4.  I / O转换：Beam带有许多“IO” - 库PTransforms可以读取或写入各种外部存储系统的数据。

典型的Beam驱动程序的工作原理如下：

1.  创建Pipeline对象并设置管道执行选项，包括Pipeline Runner。
2.  为管道数据创建初始PCollection，使用IO从外部存储系统读取数据，或使用Create转换从内存数据构建PCollection。
3.  将PTransforms应用于每个PCollection。变换可以更改，过滤，分组，分析或以其他方式处理PCollection中的元素。转换创建新的输出PCollection而不修改输入集合。典型的管道依次将后续变换应用于每个新输出PCollection直到处理完成。但是，请注意，管道不必是一个接一个地应用的单个直线变换：将PCollections视为变量，将PTransforms视为应用于这些变量的函数：管道的形状可以是任意复杂的处理图形。
4.  使用IO将最终的转换后的PCollection写入外部源。
5.  使用指定的Pipeline Runner运行管道。

运行Beam驱动程序时，您指定的Pipeline Runner会根据您创建的PCollection对象构建管道的工作流图，并转换您已应用的对象。然后使用适当的分布式处理后端执行该图，在该后端成为异步“作业”（或等效的）

--------------------

## 创建Pipeline ##

Pipeline抽象封装了数据处理任务中的所有数据和步骤。 您的Beam驱动程序通常首先构建一个对象，然后使用该对象作为创建管道数据集的基础，如PCollections及其作为Transforms的操作。

要使用Beam，您的驱动程序必须首先创建Beam SDK类Pipeline的实例（通常在main（）函数中）。 创建管道时，还需要设置一些配置选项。 您可以以编程方式设置管道的配置选项，但通常更容易提前设置选项（或从命令行读取它们），并在创建对象时将它们传递给Pipeline对象。

// In order to start creating the pipeline for execution, a Pipeline object and a Scope object are needed.
    p, s := beam.NewPipelineWithRoot()

**配置管道选项**

使用管道选项配置管道的不同方面，例如将执行管道的管道运行程序以及所选运行程序所需的任何运行程序特定的配置。您的管道选项可能包含诸如项目ID或存储文件的位置之类的信息。

在您选择的运行器上运行管道时，您的代码将可以使用PipelineOptions的副本。例如，如果将一个PipelineOptions参数添加到DoFn的@ProcessElement方法，它将由系统填充。

从命令行参数设置PipelineOptions  
虽然您可以通过创建PipelineOptions对象并直接设置字段来配置管道，但Beam SDK包含一个命令行解析器，您可以使用该命令行解析器使用命令行参数在PipelineOptions中设置字段。

要从命令行读取选项，请构造PipelineOptions对象，如以下示例代码所示：

// If beamx or Go flags are used, flags must be parsed first.
    flag.Parse()

这解释了遵循以下格式的命令行参数：

--<option>=<value>   - <选项> = <值>

注意：附加方法.withValidation将检查所需的命令行参数并验证参数值。

以这种方式构建PipelineOptions允许您将任何选项指定为命令行参数。

注意：[WordCount example pipeline][]示例管道演示了如何使用命令行选项在运行时设置管道选项。

创建自定义选项  
除标准PipelineOptions外，您还可以添加自己的自定义选项。 要添加自己的选项，请为每个选项定义具有getter和setter方法的接口，如以下示例所示：

var (
      input = flag.String("input", "", "")
      output = flag.String("output", "", "")
    )

您还可以指定描述，该描述在用户将--help作为命令行参数传递时显示，并显示默认值。

您可以使用注释设置描述和默认值，如下所示：

var (
      input = flag.String("input", "gs://my-bucket/input", "File(s) to read.")
      output = flag.String("output", "gs://my-bucket/output", "Output file.")
    )

现在您的管道可以接受--myCustomOption = value作为命令行参数。

--------------------

## PCollections ##

抽象表示潜在分布的多元素数据集。 您可以将PCollection视为“管道”数据; Beam变换使用PCollection对象作为输入和输出。 因此，如果要处理管道中的数据，则必须采用PCollection的形式。

在创建管道之后，您需要首先以某种形式创建至少一个PCollection。 您创建的PCollection用作管道中第一个操作的输入。

**创建PCollection**  
您可以通过使用Beam的Source API从外部源读取数据来创建PCollection，也可以在驱动程序中创建存储在内存中集合类中的数据的PCollection。 前者通常是生产管道如何获取数据; Beam的Source API包含适配器，可帮助您从外部源读取大型基于云的文件，数据库或订阅服务。 后者主要用于测试和调试目的。

从外部来源阅读  
要从外部源读取，请使用Beam提供的I / O适配器之一。 适配器的确切用法各不相同，但它们都从一些外部数据源读取并返回一个PCollection，其元素代表该源中的数据记录。

每个数据源适配器都有一个Read变换; 要阅读，您必须将该变换应用于Pipeline对象本身。 例如，从外部文本文件读取并返回其元素为String类型的PCollection，每个String表示文本文件中的一行。 以下是您应用于管道创建PCollection的方法：

lines := textio.Read(s, "protocol://path/to/some/inputData.txt")

请参阅有关I / O的部分，以了解有关如何从Beam SDK支持的各种数据源进行读取的更多信息。

从内存数据创建PCollection

以下示例代码显示如何从内存中创建PCollection：

counted := CountWords(s, lines)

**PCollection特性**

PCollection由为其创建的特定Pipeline对象拥有; 多个管道无法共享PCollection。 在某些方面，PCollection的功能类似于集合类。 但是，PCollection可以在几个关键方面有所不同：

元素类型  
PCollection的元素可以是任何类型，但必须都是相同的类型。 但是，为了支持分布式处理，Beam需要能够将每个单独的元素编码为字节串（因此元素可以传递给分布式工作者）。 Beam SDK提供了一种数据编码机制，包括常用类型的内置编码，以及支持根据需要指定自定义编码。

不变性  
PCollection是不可变的。 创建后，您无法添加，删除或更改单个元素。 Beam Transform可以处理PCollection的每个元素并生成新的管道数据（作为新的PCollection），但它不会消耗或修改原始输入集合。

随机访问  
PCollection不支持对单个元素的随机访问。相反，Beam Transforms会单独考虑PCollection中的每个元素。

大小和有界  
PCollection是一个庞大的，不可改变的“包”元素。 PCollection可以包含的元素数量没有上限;任何给定的PCollection可能适合单个机器上的内存，或者它可能代表由持久数据存储支持的非常大的分布式数据集.

PCollection的大小可以是有界的，也可以是无界的。有界PCollection表示已知固定大小的数据集，而无界PCollection表示无限大小的数据集。 PCollection是有界还是无界取决于它所代表的数据集的来源。从批处理数据源（例如文件或数据库）读取会创建有界PCollection。从流式传输或不断更新的数据源（如Pub / Sub或Kafka）中读取，会创建一个无界的PCollection（除非您明确告诉它不要）。

PCollection的有界（或无界）性质会影响Beam处理数据的方式。可以使用批处理作业处理有界PCollection，批处理作业可以读取整个数据集一次，并在有限长度的作业中执行处理。必须使用连续运行的流式作业处理无界PCollection，因为整个集合在任何时候都无法进行处理。

Beam使用窗口将连续更新的无界PCollection划分为有限大小的逻辑窗口。这些逻辑窗口由与数据元素相关联的某些特征确定，例如时间戳。聚合转换（例如GroupByKey和Combine）基于每个窗口工作 - 当生成数据集时，它们将每个PCollection作为这些有限窗口的连续处理。

元素时间戳  
PCollection中的每个元素都有一个关联的固有时间戳。每个元素的时间戳最初由创建PCollection的Source分配。创建无界PCollection的源通常会为每个新元素分配一个时间戳，该时间戳对应于读取或添加元素的时间。

注意：为固定数据集创建有界PCollection的源也会自动分配时间戳，但最常见的行为是为每个元素分配相同的时间戳（Long.MIN\_VALUE）。

时间戳对于包含具有固有时间概念的元素的PCollection非常有用。如果您的管道正在读取事件流（如推文或其他社交媒体消息），则每个元素可能会使用事件发布的时间作为元素时间戳。

如果源不为您执行此操作，您可以手动将时间戳分配给PCollection的元素。如果元素具有固有的时间戳，则需要执行此操作，但时间戳位于元素本身的结构中（例如服务器日志条目中的“time”字段）。 Beam具有将PCollection作为输入的变换，并输出附加了时间戳的相同PCollection;有关如何执行此操作的详细信息，请参阅[Adding Timestamps][]。

--------------------

## Transforms ##

Transforms是管道中的操作，并提供通用的处理框架。您以函数对象的形式提供处理逻辑（通俗地称为“用户代码”），并且您的用户代码应用于输入PCollection（或多个PCollection）的每个元素。根据您选择的管道运行程序和后端，群集中的许多不同工作程序可以并行执行用户代码的实例。在每个worker上运行的用户代码生成输出元素，这些元素最终被添加到转换产生的最终输出PCollection中。

Beam SDK包含许多不同的变换，您可以将这些变换应用于管道的PCollections。这些包括通用核心转换，例如ParDo或Combine。 SDK中还包含预先编写的复合变换，它们以一种有用的处理模式组合一个或多个核心变换，例如计算或组合集合中的元素。您还可以定义自己更复杂的复合变换，以适应管道的确切用例。

**应用变换**

要调用转换，必须将其应用于输入PCollection。 Beam SDK中的每个变换都有一个通用的apply方法。调用多个Beam变换类似于方法链接，但略有不同：将变换应用于输入PCollection，将变换本身作为参数传递，操作返回输出PCollection。这采取一般形式：

因为Beam对PCollection使用通用的apply方法，所以您既可以顺序链变换，也可以应用包含嵌套在其中的其他变换的变换（在Beam SDK中称为复合变换）。

如何应用管道的转换决定了管道的结构。考虑管道的最佳方法是作为有向非循环图，其中节点是PCollections，边是变换。例如，您可以链变换以创建顺序管道，如下所示：

![20190214205754772.png][]

图：具有三个顺序变换的线性流水线。

但是，请注意，转换不会消耗或以其他方式更改输入集合 - 请记住，PCollection根据定义是不可变的。 这意味着您可以将多个转换应用于相同的输入PCollection以创建分支管道，如下所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70][]

图：分支管道。 两个转换应用于数据库表行的单个PCollection。

您还可以构建自己的复合变换，在单个较大的变换中嵌套多个子步骤。 复合变换对于构建可在许多不同位置使用的可重复使用的简单步骤序列特别有用。

**Beam核心变换**  
Beam提供以下核心转换，每个转换代表不同的处理范例：

*  `ParDo`
 *  `GroupByKey`
 *  `CoGroupByKey`
 *  `Combine`
 *  `Flatten`
 *  `Partition`

ParDo

ParDo是用于通用并行处理的Beam变换。 ParDo处理范例类似于Map / Shuffle / Reduce风格算法的“Map”阶段：ParDo变换考虑输入PCollection中的每个元素，对该元素执行一些处理功能（您的用户代码），并发出零，一个或多个元素到输出PCollection。

ParDo对于各种常见的数据处理操作非常有用，包括：

1.  过滤数据集: 您可以使用ParDo考虑PCollection中的每个元素，并将该元素输出到新集合，或将其丢弃。
2.  格式化或类型转换数据集中的每个元素: 如果输入的PCollection包含的元素类型或格式与您想要的不同，则可以使用ParDo对每个元素执行转换，并将结果输出到新的PCollection。
3.  提取数据集中每个元素的部分: 例如，如果您具有包含多个字段的记录的PCollection，则可以使用ParDo将要考虑的字段解析为新的PCollection。
4.  对数据集中的每个元素执行计算: 您可以使用ParDo对PCollection的每个元素或某些元素执行简单或复杂的计算，并将结果输出为新的PCollection。

在这些角色中，ParDo是管道中常见的中间步骤。您可以使用它从一组原始输入记录中提取某些字段，或将原始输入转换为其他格式;您也可以使用ParDo将处理后的数据转换为适合输出的格式，如数据库表行或可打印字符串。

应用ParDo转换时，您需要以DoFn对象的形式提供用户代码。 DoFn是一个Beam SDK类，它定义了分布式处理函数。在创建DoFn的子类时，请注意您的子类应遵循为Beam变换编写用户代码的要求。

Applying ParDo

与所有Beam变换一样，您通过在输入PCollection上调用apply方法并将ParDo作为参数传递来应用ParDo，如以下示例代码所示：

// words is the input PCollection of strings
    var words beam.PCollection = ...
    
    func computeWordLengthFn(word string) int {
          return len(word)
    }
    
    wordLengths := beam.ParDo(s, computeWordLengthFn, words)

在示例中，我们的输入PCollection包含String值。 我们应用ParDo变换来指定一个函数（ComputeWordLengthFn）来计算每个字符串的长度，并将结果输出到一个新的PCollection of Integer值，用于存储每个单词的长度。

Creating a DoFn

传递给ParDo的DoFn对象包含应用于输入集合中的元素的处理逻辑。 当您使用Beam时，您将编写的最重要的代码片段通常是这些DoFns - 它们定义了管道的确切数据处理任务。

注意：创建DoFn时，请注意为Beam变换编写用户代码的要求[(Requirements for writing user code for Beam transforms][Requirements for writing user code for Beam transforms] )，并确保代码遵循它们。

通常会调用给定的DoFn实例一次或多次来处理某些任意元素束。 但是，Beam不保证确切的调用次数; 可以在给定的工作节点上多次调用它以解决故障和重试。 因此，您可以在处理方法的多个调用之间缓存信息，但是如果这样做，请确保实现不依赖于调用的数量。

在您的处理方法中，您还需要满足一些不变性要求，以确保Beam和处理后端可以安全地序列化和缓存管道中的值。 您的方法应符合以下要求：

Lightweight DoFns and other abstractions

如果您的功能相对简单，您可以通过提供内嵌的轻量级DoFn来简化ParDo的使用，如。

这是前面的例子，ParDo with ComputeLengthWordsFn，DoFn指定为：

// words is the input PCollection of strings
    var words beam.PCollection = ...
    
    lengths := beam.ParDo(s, func (word string) int {
          return len(word)
    }, words)

如果您的ParDo执行输入元素与输出元素的一对一映射 - 也就是说，对于每个输入元素，它应用一个只生成一个输出元素的函数，则可以使用更高级别的转换。  
这是前一个使用的示例：

GroupByKey

这是一个并行缩减操作，类似于Map / Shuffle / Reduce风格算法的Shuffle阶段。 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表示从多图（多个键到单个值）到uni-map（值集合的唯一键）的变换。

GroupByKey and unbounded PCollections

如果使用无界PCollections，则必须使用非全局窗口或聚合触发器才能执行GroupByKey或CoGroupByKey。这是因为有界GroupByKey或CoGroupByKey必须等待收集某个键的所有数据，但是对于无界集合，数据是无限的。窗口化和/或触发器允许分组在无界数据流内的逻辑有限数据束上操作。

如果您将GroupByKey或CoGroupByKey应用于一组无界PCollections而不为每个集合设置非全局窗口策略，触发策略或两者，则Beam会在管道构建时生成IllegalStateException错误。

使用GroupByKey或CoGroupByKey对应用了窗口策略的PCollection进行分组时，要分组的所有PCollections必须使用相同的窗口策略和窗口大小调整。例如，您要合并的所有集合必须使用（假设）相同的5分钟固定窗口，或每30秒开始的4分钟滑动窗口。

如果您的管道尝试使用GroupByKey或CoGroupByKey将PCollections与不兼容的窗口合并，则Beam会在管道构建时生成IllegalStateException错误。

CoGroupByKey

CoGroupByKey执行具有相同密钥类型的两个或多个键/值PCollections的关系连接。设计您的管道( [Design Your Pipeline][])显示使用连接的示例管道。

如果您有多个数据集来提供有关相关事物的信息，请考虑使用CoGroupByKey。例如，假设您有两个不同的用户数据文件：一个文件有名称和电子邮件地址;另一个文件有姓名和电话号码。您可以使用用户名作为公共密钥并将其他数据作为关联值来加入这两个数据集。加入后，您有一个数据集，其中包含与每个名称关联的所有信息（电子邮件地址和电话号码）。

如果使用无界PCollections，则必须使用非全局窗口([non-global windowing][] )或聚合触发器([aggregation trigger][])才能执行CoGroupByKey。有关详细信息，请参阅GroupByKey和无界PCollections([GroupByKey and unbounded PCollections][])。

以下概念示例使用两个输入集合来显示CoGroupByKey的机制。

在CoGroupByKey之后，结果数据包含与来自任何输入集合的每个唯一键相关联的所有数据。

final TupleTag<String> emailsTag = new TupleTag<>();
    final TupleTag<String> phonesTag = new TupleTag<>();
    
    final List<KV<String, CoGbkResult>> expectedResults =
        Arrays.asList(
            KV.of(
                "amy",
                CoGbkResult.of(emailsTag, Arrays.asList("amy@example.com"))
                    .and(phonesTag, Arrays.asList("111-222-3333", "333-444-5555"))),
            KV.of(
                "carl",
                CoGbkResult.of(emailsTag, Arrays.asList("carl@email.com", "carl@example.com"))
                    .and(phonesTag, Arrays.asList("444-555-6666"))),
            KV.of(
                "james",
                CoGbkResult.of(emailsTag, Arrays.asList())
                    .and(phonesTag, Arrays.asList("222-333-4444"))),
            KV.of(
                "julia",
                CoGbkResult.of(emailsTag, Arrays.asList("julia@example.com"))
                    .and(phonesTag, Arrays.asList())));

以下代码示例将两个PCollections与CoGroupByKey连接，然后使用ParDo来使用结果。然后，代码使用标签来查找和格式化每个集合中的数据。

PCollection<KV<String, CoGbkResult>> results =
        KeyedPCollectionTuple.of(emailsTag, emails)
            .and(phonesTag, phones)
            .apply(CoGroupByKey.create());
    
    PCollection<String> contactLines =
        results.apply(
            ParDo.of(
                new DoFn<KV<String, CoGbkResult>, String>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    KV<String, CoGbkResult> e = c.element();
                    String name = e.getKey();
                    Iterable<String> emailsIter = e.getValue().getAll(emailsTag);
                    Iterable<String> phonesIter = e.getValue().getAll(phonesTag);
                    String formattedResult =
                        Snippets.formatCoGbkResults(name, emailsIter, phonesIter);
                    c.output(formattedResult);
                  }
                }));

格式化的数据如下所示：

final List<String> formattedResults =
        Arrays.asList(
            "amy; ['amy@example.com']; ['111-222-3333', '333-444-5555']",
            "carl; ['carl@email.com', 'carl@example.com']; ['444-555-6666']",
            "james; []; ['222-333-4444']",
            "julia; ['julia@example.com']; []");

Combine

是一个Beam变换，用于组合数据中的元素或值集合。 Combine具有适用于整个PCollections的变体，还有一些变量组合了PCollections键/值对中每个键的值。

应用组合变换时，必须提供包含组合元素或值的逻辑的函数。组合函数应该是可交换的和关联的，因为函数不一定只对具有给定键的所有值调用一次。因为输入数据（包括值集合）可以分布在多个工作者之间，所以可以多次调用组合函数以对值集合的子集执行部分组合。 Beam SDK还为常见的数字组合操作（如sum，min和max）提供了一些预构建的组合函数。

简单的组合操作（例如总和）通常可以实现为简单的功能。更复杂的组合操作可能需要您创建CombineFn的子类，其具有与输入/输出类型不同的累积类型。

Simple combinations using simple functions

以下示例代码显示了一个简单的组合函数。

// 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;
      }
    }

Advanced combinations using CombineFn

对于更复杂的组合函数，您可以定义CombineFn的子类。 如果组合功能需要更复杂的累加器，则必须使用CombineFn，必须执行其他预处理或后处理，可能会更改输出类型或考虑密钥。

一般的组合操作包括四个操作。 在创建CombineFn的子类时，必须通过重写相应的方法来提供四个操作：

1.  Create Accumulator创建一个新的“本地”累加器。在示例情况下，取平均值，本地累加器跟踪值的运行总和（我们的最终平均除法的分子值）和到目前为止总和的值的数量（分母值）。它可以以分布式方式被调用任意次数。
2.  Add Input将一个输入元素添加到累加器，返回累加器值。在我们的示例中，它将更新总和并增加计数。它也可以并行调用。
3.  Merge Accumulators将几个累加器合并为一个累加器;这是在最终计算之前将多个累加器中的数据组合的方式。在平均值计算的情况下，表示除法的每个部分的累加器被合并在一起。可以在其输出上再次调用它。
4.  Extract Output提取输出执行最终计算。在计算平均值的情况下，这意味着将所有值的组合总和除以求和的值的数量。它在最终的合并累加器上调用一次。

以下示例代码显示如何定义计算平均值的CombineFn：

如果要组合键值对的PCollection，则每键组合通常就足够了。 如果您需要基于密钥更改组合策略（例如，某些用户为MIN，其他用户为MAX），则可以定义KeyedCombineFn以访问组合策略中的密钥。

Combining a PCollection into a single value

使用全局组合将给定PCollection中的所有元素转换为单个值，在管道中表示为包含一个元素的新PCollection。 以下示例代码显示如何应用Beam提供的求和组合函数，以便为整数的PCollection生成单个和值。

Combine and global windowing

如果输入PCollection使用默认的全局窗口，则默认行为是返回包含一个项目的PCollection。 该项的值来自您在应用Combine时指定的组合函数中的累加器。 例如，Beam提供的sum组合函数返回零值（空输入的总和），而min组合函数返回最大值或无限值。

要使Combine在输入为空时返回空PCollection，请在应用Combine变换时指定.withoutDefaults，如下面的代码示例所示：

Combine and non-global windowing

如果您的PCollection使用任何非全局窗口函数，则Beam不提供默认行为。 应用Combine时，您必须指定以下选项之一：

指定.withoutDefaults，其中输入PCollection中为空的窗口在输出集合中同样为空。  
指定.asSingletonView，其中输出立即转换为PCollectionView，当用作侧输入时，它将为每个空窗口提供默认值。 如果管道的Combine的结果将在管道中稍后用作侧输入，则通常只需要使用此选项。

Combining values in a keyed PCollection

在创建键控PCollection之后（例如，通过使用GroupByKey变换），常见模式是将与每个键关联的值集合组合成单个合并值。 使用GroupByKey中的上一个示例，一个名为groupedWords的键分组PCollection如下所示：

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的子类。

Flatten

是用于存储相同数据类型的PCollection对象的Beam变换。 Flatten将多个PCollection对象合并为一个逻辑PCollection。

以下示例显示如何应用Flatten变换以合并多个PCollection对象。

Data encoding in merged collections

默认情况下，输出PCollection的编码器与输入PCollectionList中第一个PCollection的编码器相同。 但是，输入的PCollection对象可以使用不同的编码器，只要它们都包含所选语言的相同数据类型即可。

Merging windowed collections

使用Flatten合并应用了窗口策略的PCollection对象时，要合并的所有PCollection对象必须使用兼容的窗口策略和窗口大小调整。 例如，您正在合并的所有集合必须全部使用（假设）相同的5分钟固定窗口或每30秒开始的4分钟滑动窗口。

如果您的管道尝试使用Flatten将PCollection对象与不兼容的窗口合并，则在构建管道时，Beam会生成IllegalStateException错误。

Partition

是用于存储相同数据类型的PCollection对象的Beam变换。 分区将单个PCollection拆分为固定数量的较小集合。

分区根据您提供的分区函数划分PCollection的元素。 分区函数包含确定如何将输入PCollection的元素拆分为每个结果分区PCollection的逻辑。 必须在图形构建时确定分区数。 例如，您可以在运行时将分区数作为命令行选项传递（然后将用于构建管道图），但是无法确定管道中间的分区数（基于之后计算的数据） 例如，构建了您的管道图。

以下示例将PCollection划分为百分位数组。

**为Beam变换编写用户代码的要求**

为Beam转换构建用户代码时，应记住执行的分布式特性。例如，您可能会在许多不同的计算机上并行运行许多功能副本，这些副本可以独立运行，而无需与任何其他副本通信或共享状态。根据您为管道选择的管道运行器和处理后端，可以重试或运行多次用户代码功能副本。因此，您应该谨慎地在用户代码中包含状态依赖等内容。

通常，您的用户代码必须至少满足以下要求：

1.  您的函数对象必须是可序列化的。
2.  您的函数对象必须是线程兼容的，并且请注意Beam SDK不是线程安全的。

此外，建议您使功能对象具有幂等性。 Beam支持非幂等函数，但在有外部副作用时需要额外考虑以确保正确性。

注意：这些要求适用于DoFn（与ParDo变换一起使用的函数对象），CombineFn（与Combine变换一起使用的函数对象）和WindowFn（与Window变换一起使用的函数对象）的子类。

Serializability

您为变换提供的任何函数对象都必须是完全可序列化的。这是因为需要将函数的副本序列化并传输到处理集群中的远程工作程序。用户代码的基类，如DoFn，CombineFn和WindowFn，已实现Serializable;但是，您的子类不能添加任何不可序列化的成员。

您应该记住的一些其他可串行化因素是：

1.  函数对象中的瞬态字段不会传输到工作线实例，因为它们不会自动序列化。
2.  在序列化之前避免加载包含大量数据的字段。
3.  函数对象的各个实例无法共享数据。
4.  在应用函数对象后对其进行变换将不起作用。
5.  使用匿名内部类实例在内联声明函数对象时要小心。在非静态上下文中，内部类实例将隐式包含指向封闭类和该类的状态的指针。封闭类也将被序列化，因此适用于函数对象本身的相同注意事项也适用于此外部类。

Thread-compatibility

您的函数对象应该是线程兼容的。 除非您明确创建自己的线程，否则函数对象的每个实例都会在工作实例上由一个线程一次访问。 但请注意，Beam SDK不是线程安全的。 如果在用户代码中创建自己的线程，则必须提供自己的同步。 请注意，函数对象中的静态成员不会传递给工作器实例，并且可以从不同的线程访问函数的多个实例。

Idempotence

建议您使您的函数对象具有幂等性 - 也就是说，它可以根据需要重复或重试，而不会导致意外的副作用。 支持非幂等函数，但Beam模型不保证可以调用或重试用户代码的次数; 因此，保持函数对象的幂等性可以使管道输出具有确定性，并且您的变换行为更易于预测且更易于调试。

**侧输入Side inputs**

除了主输入PCollection之外，您还可以以侧输入的形式为ParDo变换提供额外的输入。 侧输入是DoFn每次处理输入PCollection中的元素时可以访问的附加输入。 指定侧输入时，可以创建一个其他数据的视图，这些数据可以在处理每个元素时从ParDo变换的DoFn中读取。

如果ParDo在处理输入PCollection中的每个元素时需要注入额外数据，则侧输入很有用，但需要在运行时确定附加数据（而不是硬编码）。 这些值可能由输入数据确定，或取决于管道的不同分支。

Passing side inputs to ParDo

Side inputs and windowing

窗口化PCollection可以是无限的，因此不能压缩成单个值（或单个集合类）。当您创建窗口PCollection的PCollectionView时，PCollectionView表示每个窗口一个实体（每个窗口一个单独，每个窗口一个列表等）。

Beam使用主输入元素的窗口来查找侧输入元素的适当窗口。 Beam将主输入元素的窗口投影到侧输入的窗口集中，然后使用结果窗口中的侧输入。如果主输入和侧输入具有相同的窗口，则投影提供精确的相应窗口。但是，如果输入具有不同的窗口，则Beam使用投影选择最合适的侧输入窗口。

例如，如果使用一分钟的固定时间窗口对主输入进行窗口化，并且使用一小时的固定时间窗口对侧面输入进行窗口化，则Beam将主输入窗口投影到侧面输入窗口设置并选择侧面输入来自适当的小时长侧输入窗口的值。

如果主输入元素存在于多个窗口中，则会多次调用processElement，每个窗口一次。每次调用processElement都会为主输入元素投射“当前”窗口，因此每次都可能提供不同的侧输入视图。

如果侧输入有多个触发器触发，则Beam使用最新触发器触发的值。如果您使用具有单个全局窗口的侧输入并指定触发器，则此功能特别有用。

**额外产出 Additional outputs**

虽然ParDo始终生成主输出PCollection（作为应用的返回值），但您也可以让ParDo生成任意数量的其他输出PCollections。 如果您选择具有多个输出，则ParDo将返回捆绑在一起的所有输出PCollections（包括主输出）。

Tags for multiple outputs

Emitting to multiple outputs in your DoFn

**复合变换Composite transforms**

变换可以具有嵌套结构，其中复杂变换执行多个更简单的变换（例如多个ParDo，Combine，GroupByKey或甚至其他复合变换）。 这些变换称为复合变换。 在单个复合变换中嵌套多个变换可以使您的代码更加模块化，更易于理解。

Beam SDK包含许多有用的复合变换。 有关转换列表，请参阅API参考页面：

*  [Pre-written Beam transforms for Java][]
 *  [Pre-written Beam transforms for Python][]

An example composite transform

WordCount示例程序中的CountWords变换是复合变换的示例（[WordCount example program][]）。 CountWords是一个PTransform子类，由多个嵌套转换组成。

在其expand方法中，CountWords转换应用以下转换操作：

1.  它在文本行的输入PCollection上应用ParDo，产生单个单词的输出PCollection。
2.  它将Beam SDK库转换Count应用于PCollection的单词，生成键/值对的PCollection。 每个键代表文本中的一个单词，每个值表示单词出现在原始数据中的次数。

请注意，这也是嵌套复合变换的一个示例，因为Count本身就是一个复合变换。

复合变换的参数和返回值必须与整个变换的初始输入类型和最终返回类型匹配，即使变换的中间数据多次更改类型也是如此。  
WordCount shìlì chéngxù zhō

Creating a composite transform

要创建自己的复合变换，请创建PTransform类的子类并覆盖expand方法以指定实际的处理逻辑。然后，您可以像使用Beam SDK中的内置转换一样使用此转换。

以下代码示例演示如何声明接受输入的字符串的PCollection的PTransform，并输出整数的PCollection：

在PTransform子类中，您需要覆盖expand方法。您可以在expand方法中添加PTransform的处理逻辑。您对expand的覆盖必须接受相应类型的输入PCollection作为参数，并将输出PCollection指定为返回值。

以下代码示例显示如何覆盖上一个示例中声明的ComputeWordLengths类的展开：

只要覆盖PTransform子类中的expand方法以接受相应的输入PCollection并返回相应的输出PCollection，您就可以包含任意数量的转换。这些转换可以包括核心转换，复合转换或Beam SDK库中包含的转换。

注意：PTransform的expand方法不应由转换用户直接调用。相反，您应该在PCollection本身上调用apply方法，并将transform作为参数。这允许转换嵌套在管道的结构中。

PTransform Style Guide

PTransform样式指南（[PTransform Style Guide][]）包含此处未包含的其他信息，例如样式指南，日志记录和测试指南以及特定于语言的注意事项。 当您想要编写新的复合PTransforms时，该指南是一个有用的起点。

--------------------

## Pipeline I/O ##

创建管道时，通常需要从某些外部源（例如文件或数据库）读取数据。 同样，您可能希望管道将其结果数据输出到外部存储系统。 Beam为许多常见的数据存储类型提供读写转换（[number of common data storage types][].）。 如果希望管道读取或写入内置转换不支持的数据存储格式，则可以实现自己的读写转换（[implement your own read and write transforms][].）。

**Reading input data**

读取转换从外部源读取数据并返回数据的PCollection表示以供管道使用。 您可以在构建管道时随时使用读取转换来创建新的PCollection，尽管它在管道开始时最常见。

**Writing output data**

写入转换将PCollection中的数据写入外部数据源。 您最常使用管道末尾的写入转换来输出管道的最终结果。 但是，您可以使用写入转换在管道中的任何位置输出PCollection的数据。

**File-based input and output data**

Reading from multiple locations

许多读变换支持从与您提供的glob运算符匹配的多个输入文件中读取。 请注意，glob运算符是特定于文件系统的，并遵循特定于文件系统的一致性模型。 以下TextIO示例使用glob运算符（\*）来读取在给定位置具有前缀“input-”和后缀“.csv”的所有匹配输入文件：

要将来自不同源的数据读入单个PCollection，请单独读取每个PCollection，然后使用Flatten变换创建单个PCollection。

Writing to multiple output files

对于基于文件的输出数据，默认情况下，写入转换为多个输出文件。 将输出文件名传递给写入转换时，文件名将用作写入转换生成的所有输出文件的前缀。 您可以通过指定后缀为每个输出文件追加后缀。

以下写入转换示例将多个输出文件写入某个位置。 每个文件都有前缀“numbers”，一个数字标签和后缀“.csv”。

**Beam-provided I/O transforms**

有关当前可用I / O转换的列表（[Beam-provided I/O Transforms][number of common data storage types]），请参见Beam-provided I / O Transforms页面。

--------------------

## 数据编码和类型安全 Data encoding and type safety ##

当Beam运行程序执行您的管道时，它们通常需要实现PCollections中的中间数据，这需要将元素转换为字节字符串和从字节字符串转换元素。 Beam SDK使用称为Coders的对象来描述如何编码和解码给定PCollection的元素。

请注意，在与外部数据源或接收器交互时，编码器与解析或格式化数据无关。 通常应使用ParDo或MapElements等转换明确地进行此类解析或格式化。

请注意，编码器与类型不一定具有1：1的关系。 例如，Integer类型可以有多个有效的编码器，输入和输出数据可以使用不同的Integer编码器。 转换可能包含使用BigEndianIntegerCoder的Integer类型输入数据和使用VarIntCoder的Integer类型输出数据。

**Specifying coders**

Beam SDK需要为您的管道中的每个PCollection提供一个编码器。 在大多数情况下，Beam SDK能够根据其元素类型或产生它的变换自动推断出PCollection的编码器，但是，在某些情况下，管道作者需要明确指定编码器，或者开发编码器以进行编码。 他们的自定义类型

当尝试自动推断出PCollection的编码器时，Beam SDK使用各种机制。

注意：如果使用“创建”变换从内存数据创建PCollection，则不能依赖编码器推断和默认编码器。 Create无法访问其参数的任何键入信息，如果参数列表包含其确切的运行时类没有注册默认编码器的值，则可能无法推断编码器。

**Default coders and the CoderRegistry**

每个Pipeline对象都有一个CoderRegistry对象，它将语言类型映射到管道应该用于这些类型的默认编码器。 您可以自己使用CoderRegistry查找给定类型的默认编码器，或者为给定类型注册新的默认编码器。

CoderRegistry包含使用Beam SDK创建的任何管道的编码器到标准类型的默认映射。 下表显示了标准映射

Looking up a default coder

Setting the default coder for a type

要为特定管道的类型设置默认编码器，您需要获取并修改管道的CoderRegistry。 您可以使用该方法获取CoderRegistry对象，然后使用该方法为目标类型注册新的Coder。

以下示例代码演示了如何为管道的值设置默认Coder，在本例中为BigEndianIntegerCoder

Annotating a custom data type with a default coder

--------------------

## Windowing ##

窗口化根据其各个元素的时间戳细分PCollection。 聚合多个元素的变换（例如GroupByKey和Combine）在每个窗口的基础上隐式工作 - 它们将每个PCollection作为一系列多个有限窗口处理，尽管整个集合本身可能具有无限大小。

一个相关的概念，称为触发器，确定何时在无界数据到达时发出聚合结果。 您可以使用触发器来优化PCollection的窗口策略。 触发器允许您处理迟到的数据或提供早期结果。 有关更多信息，请参阅触发器部分（ [triggers][aggregation trigger] ）。

**Windowing basics**

一些Beam变换（例如GroupByKey和Combine）通过公共密钥对多个元素进行分组。通常，该分组操作将所有在整个数据集中具有相同密钥的元素分组。利用无界数据集，不可能收集所有元素，因为新元素不断被添加并且可能无限多（例如流数据）。如果您正在使用无界PCollections，则窗口化尤其有用。

在Beam模型中，任何PCollection（包括无界PCollections）都可以细分为逻辑窗口。根据PCollection的窗口函数将PCollection中的每个元素分配给一个或多个窗口，并且每个单独的窗口包含有限数量的元素。然后，对变换进行分组，然后在每个窗口的基础上考虑每个PCollection的元素。例如，GroupByKey通过键和窗口隐式地对PCollection的元素进行分组。

注意：Beam的默认窗口行为是将PCollection的所有元素分配给单个全局窗口并丢弃后期数据，即使对于无界PCollections也是如此。在无界PCollection上使用GroupByKey之类的分组转换之前，必须至少执行以下操作之一：

1.  设置非全局窗口函数。 请参阅设置PCollection的窗口功能（ [Setting your PCollection’s windowing function][non-global windowing].）。
2.  设置非默认触发器。 这允许全局窗口在其他条件下发出结果，因为默认的窗口行为（等待所有数据到达）将永远不会发生。

如果没有为无界PCollection设置非全局窗口函数或非默认触发器，并随后使用GroupByKey或Combine等分组转换，则管道将在构造时生成错误，您的作业将失败。

窗口约束Windowing constraints

为PCollection设置窗口函数后，下次将分组变换应用于该PCollection时将使用元素的窗口。 窗口分组根据需要进行。 如果使用Window变换设置窗口函数，则会将每个元素分配给一个窗口，但只有在GroupByKey或Combine跨窗口和键聚合时才会考虑窗口。 这可能会对您的管道产生不同的影响。 考虑下图中的示例管道：

![20190215213923612.png][]

图：管道应用窗口

在上面的管道中，我们通过使用KafkaIO读取一组键/值对来创建无界PCollection，然后使用Window变换将窗口函数应用于该集合。 然后，我们将ParDo应用于集合，然后使用GroupByKey对该ParDo的结果进行分组。 窗口函数对ParDo转换没有影响，因为在GroupByKey需要窗口之前，实际上并未使用窗口。 但是，后续转换将应用于GroupByKey的结果 - 数据按键和窗口分组。

使用有界PCollections进行窗口化 Windowing with bounded PCollections

您可以在有界PCollections中使用具有固定大小数据集的窗口。 但是，请注意，窗口仅考虑附加到PCollection的每个元素的隐式时间戳，并且创建固定数据集（例如TextIO）的数据源为每个元素分配相同的时间戳。 这意味着默认情况下，所有元素都是单个全局窗口的一部分。

要使用具有固定数据集的窗口，您可以为每个元素分配自己的时间戳。 要为元素分配时间戳，请使用带有DoFn的ParDo变换，该DoFn使用新的时间戳输出每个元素（例如，Beam SDK for Java中的WithTimestamps变换）。

为了说明有界PCollection的窗口化如何影响管道处理数据的方式，请考虑以下管道：

![20190219195341174.png][]

ParDo applies once per unique key

图：在有界集合上没有窗口的GroupByKey和ParDo。

在上面的管道中，我们通过使用TextIO读取一组键/值对来创建有界PCollection。 然后，我们使用GroupByKey对集合进行分组，并将ParDo变换应用于分组的PCollection。 在此示例中，GroupByKey创建一组唯一键，然后ParDo每个键只应用一次。

请注意，即使您没有设置窗口函数，仍然有一个窗口 - 您的PCollection中的所有元素都分配给一个全局窗口。

现在，考虑相同的管道，但使用窗口函数：

![20190219195725490.png][]

ParDo applies once per unique key, in each window

图：具有窗口的GroupByKey和ParDo，在有界集合上。

和以前一样，管道创建了键/值对的有界PCollection。 然后，我们为该PCollection设置窗口函数。 GroupByKey变换基于窗口函数通过键和窗口对PCollection的元素进行分组。 随后的ParDo变换每个键应用多次，每个窗口一次。

**Provided windowing functions**

您可以定义不同类型的窗口来划分PCollection的元素。 Beam提供多种窗口功能，包括：

1.  固定时间窗口
2.  滑动时间Windows
3.  每会话Windows
4.  单一全局窗口
5.  基于日历的Windows（Beam SDK for Python不支持）

如果您有更复杂的需求，也可以定义自己的WindowFn。

请注意，每个元素在逻辑上可以属于多个窗口，具体取决于您使用的窗口函数。 例如，滑动时间窗口创建重叠窗口，其中可以将单个元素分配给多个窗口。

Fixed time windows

最简单的窗口形式是使用固定时间窗口：给定可以不断更新的带时间戳的PCollection，每个窗口可以捕获（例如）具有落入五分钟间隔的时间戳的所有元素。

固定时间窗口表示数据流中的一致持续时间，非重叠时间间隔。 考虑具有五分钟持续时间的窗口：无界PCollection中的所有元素，其中时间戳值从0:00:00到（但不包括）0:05:00属于第一个窗口，元素的时间戳值为0 ：05：00到（但不包括）0:10:00属于第二个窗口，依此类推。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 1][]

图：固定时间窗口，持续时间为30秒。

Sliding time windows

滑动时间窗口还表示数据流中的时间间隔; 但是，滑动时间窗口可以重叠。 例如，每个窗口可能会捕获五分钟的数据，但每隔十秒就会启动一个新窗口。 滑动窗口开始的频率称为周期。 因此，我们的示例将具有五分钟的窗口持续时间和十秒的时间段。

由于多个窗口重叠，因此数据集中的大多数元素将属于多个窗口。 这种窗口对于获取数据的运行平均值很有用; 使用滑动时间窗口，您可以计算过去五分钟数据的运行平均值，在我们的示例中每十秒更新一次。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 2][]

图：滑动时间窗口，窗口持续时间为1分钟，窗口周期为30秒。

Session windows

会话窗口函数定义包含在另一个元素的特定间隙持续时间内的元素的窗口。 会话窗口适用于每个密钥，对于随时间不规则分布的数据非常有用。 例如，表示用户鼠标活动的数据流可能具有散布高浓度点击的长时间空闲时间。 如果数据在最小指定间隙持续时间之后到达，则启动新窗口的开始。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 3][]

图：会话窗口，具有最小间隙持续时间。 请注意每个数据键如何根据其数据分布具有不同的窗口。

The single global window

默认情况下，PCollection中的所有数据都分配给单个全局窗口，并且后期数据将被丢弃。 如果您的数据集具有固定大小，则可以使用PCollection的全局窗口默认值。

如果使用无界数据集（例如，来自流数据源），则可以使用单个全局窗口，但在应用聚合转换（例如GroupByKey和Combine）时要小心。 具有默认触发器的单个全局窗口通常要求整个数据集在处理之前可用，这对于连续更新数据是不可能的。 要对使用全局窗口的无界PCollection执行聚合，您应该为该PCollection指定非默认触发器。

**Setting your PCollection’s windowing function**

您可以通过应用Window变换为PCollection设置窗口函数。 应用Window变换时，必须提供WindowFn。 WindowFn确定PCollection将用于后续分组变换的窗口函数，例如固定或滑动时间窗口。

设置窗口功能时，您可能还需要为PCollection设置触发器。 触发器确定何时聚合和发射每个单独的窗口，并且有助于改进窗口函数关于后期数据和计算早期结果的执行方式。 有关更多信息，请参阅触发器部分。

Fixed time windows

以下示例代码显示如何应用Window将PCollection分成固定窗口，每个窗口的长度为一分钟：

Sliding time windows

以下示例代码显示如何应用Window将PCollection划分为滑动时间窗口。 每个窗口的长度为30秒，每五秒钟开始一个新窗口：

Session windows

以下示例代码显示如何应用Window将PCollection划分为会话窗口，其中每个会话必须以至少10分钟的时间间隔分隔：

请注意，会话是按键 - 集合中的每个键都有自己的会话分组，具体取决于数据分布。

The single global window

如果您的PCollection有界（大小固定），您可以将所有元素分配到单个全局窗口。 以下示例代码显示如何为PCollection设置单个全局窗口：

**水印和后期数据Watermarks and late data**

在任何数据处理系统中，数据事件发生的时间（“事件时间”，由数据元素本身的时间戳确定）与实际数据元素在任何阶段处理的时间之间存在一定的滞后。你的管道（“处理时间”，由处理元素的系统上的时钟决定）。此外，无法保证数据事件将以与生成它们相同的顺序出现在管道中。

例如，假设我们有一个使用固定时间窗口的PCollection，窗口长达五分钟。对于每个窗口，Beam必须在给定的窗口范围内收集具有事件时间戳的所有数据（例如，在第一个窗口中的0:00和4:59之间）。时间戳超出该范围的数据（来自5:00或更晚的数据）属于不同的窗口。

但是，并不总能保证数据按时间顺序到达管道，或始终以可预测的间隔到达。光束跟踪水印，这是系统的概念，即某个窗口中的所有数据何时可以到达管道中。一旦水印越过窗口的末尾，在该窗口中带有时间戳的任何其他元素被认为是晚期数据。

从我们的示例中，假设我们有一个简单的水印，假设数据时间戳（事件时间）和数据在管道中出现的时间（处理时间）之间的滞后时间约为30秒，那么Beam将关闭第一个窗口5 ：30。如果数据记录在5:34到达，但是时间戳会将其置于0：00-4：59窗口（例如，3：38），则该记录是延迟数据。

注意：为简单起见，我们假设我们使用非常简单的水印来估计延迟时间。实际上，PCollection的数据源确定水印，水印可以更精确或更复杂。

Beam的默认窗口配置尝试确定所有数据何时到达（基于数据源的类型），然后将水印推进到窗口的末尾。此默认配置不允许延迟数据。触发器允许您修改和优化PCollection的窗口策略。您可以使用触发器来确定每个窗口何时聚合并报告其结果，包括窗口如何发出后期元素。

Managing late data

注意：Beam SDK for Python不支持管理延迟数据。

在设置PCollection的窗口策略时，可以通过调用.withAllowedLateness操作来允许延迟数据。 下面的代码示例演示了一种窗口策略，该策略允许在窗口结束后最多两天的延迟数据。

当您在PCollection上设置.withAllowedLateness时，允许延迟向前传播到从您应用的第一个PCollection派生的任何后续PCollection，允许迟到。 如果要在管道中稍后更改允许的延迟，则必须通过应用Window.configure（）。withAllowedLateness（）来明确地执行此操作。

**Adding timestamps to a PCollection’s elements**

无界源为每个元素提供时间戳。 根据您的无限制来源，您可能需要配置从原始数据流中提取时间戳的方式。

但是，有界源（例如TextIO中的文件）不提供时间戳。 如果需要时间戳，则必须将它们添加到PCollection的元素中。

您可以通过应用ParDo转换为PCollection的元素分配新的时间戳，该转换输出带有您设置的时间戳的新元素。

例如，如果您的管道从输入文件中读取日志记录，并且每个日志记录都包含一个时间戳字段; 由于管道从文件中读取记录，因此文件源不会自动分配时间戳。 您可以解析每条记录中的时间戳字段，并使用带有DoFn的ParDo转换将时间戳附加到PCollection中的每个元素。

## Triggers ##

确定何时发出每个窗口的聚合结果（称为窗格）。如果使用Beam的默认窗口配置和默认触发器，则Beam会在估计所有数据到达时输出聚合结果，并丢弃该窗口的所有后续数据。

您可以为PCollections设置触发器以更改此默认行为。 Beam提供了许多可以设置的预构建触发器：

1.  事件时间触发。这些触发器在事件时间上运行，如每个数据元素的时间戳所示。 Beam的默认触发器是基于事件时间的。
2.  处理时间触发。这些触发器在处理时间上运行 - 处理数据元素在管道中任何给定阶段的处理时间。
3.  数据驱动的触发器。这些触发器通过在数据到达每个窗口时检查数据来操作，并在数据满足某个属性时触发。目前，数据驱动的触发器仅支持在一定数量的数据元素之后触发。
4.  复合触发器。这些触发器以各种方式组合多个触发器。

在高级别，触发器提供了两个额外的功能，而不是简单地在窗口的末尾输出：

1.  触发器允许Beam在给定窗口中的所有数据到达之前发出早期结果。例如，在经过一定量的时间之后，或在一定数量的元素到达之后发射。
2.  触发器允许在事件时间水印通过窗口结束之后通过触发来处理延迟数据。

这些功能允许您根据用例控制数据流并在不同因素之间取得平衡：

1.  完整性：在计算结果之前获取所有数据有多重要？
2.  延迟：您想等待数据多长时间？ 例如，您是否等到认为自己拥有所有数据？ 你在数据到达时处理数据吗？
3.  成本：您愿意花多少计算能力/金钱来降低延迟？

例如，需要时间敏感更新的系统可能会使用严格的基于时间的触发器，该触发器每隔N秒发出一个窗口，评估数据完整性的快速性。 一个重视数据完整性而不是结果的确切时间的系统可能会选择使用Beam的默认触发器，该触发器在窗口结束时触发。

您还可以为无界PCollection设置触发器，该触发器使用单个全局窗口作为其窗口函数。 当您希望管道在无界数据集上提供定期更新时，这非常有用 - 例如，提供给当前时间的所有数据的运行平均值，每N秒更新一次或每N个元素更新一次。

**Event time triggers**

AfterWatermark触发器在事件时间运行。 根据附加到数据元素的时间戳，AfterWatermark触发器在水印通过窗口结束后发出窗口内容。 水印是一个全局进度指标，是Beam在任何给定点的管道内输入完整性的概念。 只有当水印通过窗口的末端时才会触发。

此外，您可以配置在管道在窗口结束之前或之后接收数据时触发的触发器。

以下示例显示了一种结算方案，并使用早期和晚期解雇：

Default trigger

PCollection的默认触发器基于事件时间，并在Beam的水印通过窗口结束时发出窗口的结果，然后在每次后期数据到达时触发。

但是，如果同时使用默认窗口配置和默认触发器，则默认触发器只会发出一次，并且会丢弃延迟数据。 这是因为默认窗口配置的允许延迟值为0.有关修改此行为的信息，请参阅处理延迟数据部分。

**Processing time triggers**

AfterProcessingTime触发器按处理时间运行。 例如，在接收到数据后经过一定量的处理时间后，触发器会发出一个窗口。 处理时间由系统时钟决定，而不是数据元素的时间戳。

AfterProcessingTime触发器对于从窗口触发早期结果非常有用，尤其是具有大时间范围的窗口，例如单个全局窗口。

**Data-driven triggers**

Beam提供一个数据驱动的触发器。 此触发器适用于元素计数; 它在当前窗格收集了至少N个元素后触发。 这允许窗口发出早期结果（在所有数据累积之前），如果您使用单个全局窗口，这可能特别有用。

重要的是要注意，例如，如果您指定并且只有32个元素到达，那么这32个元素将永远存在。 如果32个元素对您很重要，请考虑使用复合触发器来组合多个条件。 这允许您指定多个触发条件，例如“当我收到50个元素或每1秒时触发”。

**Setting a trigger**

使用变换为PCollection设置窗口函数时，还可以指定触发器。

Window accumulation modes

指定触发器时，还必须设置窗口的累积模式。 当触发器触发时，它会以窗格的形式发出窗口的当前内容。 由于触发器可以多次触发，因此累积模式确定系统是否在触发器触发时累积窗口窗格，或者丢弃它们。

让我们看一个使用具有固定时间窗口和基于数据的触发器的PCollection的示例。 例如，如果每个窗口代表10分钟的运行平均值，但您希望在每隔10分钟更频繁地显示UI中的平均值，则可以执行此操作。 我们假设以下条件：

PCollection使用10分钟的固定时间窗口。  
PCollection具有重复触发器，每次有3个元素到达时触发。  
下图显示了密钥X到达PCollection并分配给窗口时的数据事件。 为了使图表更简单，我们假设事件全部按顺序到达管道。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 4][]

Accumulating mode

如果我们的触发器设置为累积模式，则每次触发时触发器都会发出以下值。 请记住，每次有三个元素到达时触发器会触发：

First trigger firing:  [5, 8, 3]
     Second trigger firing: [5, 8, 3, 15, 19, 23]
     Third trigger firing:  [5, 8, 3, 15, 19, 23, 9, 13, 10]

Discarding mode

如果我们的触发器设置为丢弃模式，则触发器会在每次触发时发出以下值：

First trigger firing:  [5, 8, 3]
      Second trigger firing:           [15, 19, 23]
      Third trigger firing:                         [9, 13, 10]

Handling late data

Beam SDK for Python目前不支持允许延迟。

如果您希望管道处理水印通过窗口结束后到达的数据，则可以在设置窗口配置时应用允许的延迟。 这使您的触发器有机会对迟到的数据做出反应。 如果允许延迟设置，则默认触发器将在迟到数据到达时立即发出新结果。

设置窗口函数时，可以使用.withAllowedLateness（）设置允许的延迟：

这允许延迟传播到由于将变换应用于原始PCollection而导出的所有PCollections。 如果要在管道中稍后更改允许的延迟，可以再次显式应用Window.configure（）。withAllowedLateness（）。

**Composite triggers**

您可以组合多个触发器以形成复合触发器，并可以指定触发器以重复，最多一次或在其他自定义条件下重复发出结果。

Composite trigger types

Beam包含以下复合触发器：

1.  您可以通过.withEarlyFirings和.withLateFirings向AfterWatermark.pastEndOfWindow添加额外的早期点火或后期点火。
2.  Repeatedly.forever指定永久执行的触发器。只要满足触发条件，就会导致窗口发出结果，然后重置并重新开始。将Repeatedly.forever与.orFinally结合使用以指定导致重复触发器停止的条件会很有用。
3.  AfterEach.inOrder将多个触发器组合在一起以特定的顺序触发。每次序列中的触发器发出一个窗口时，序列前进到下一个触发器。
4.  AfterFirst接受多个触发器，并在第一次发出任何参数触发器时发出。这相当于多个触发器的逻辑OR运算。
5.  AfterAll接受多个触发器并在满足其所有参数触发器时发出。这相当于多个触发器的逻辑AND操作。
6.  orFinally可以作为最终条件，使任何触发器最后一次触发并且永远不再触发。

Composition with AfterWatermark

当Beam估计所有数据已经到达时（即当水印通过窗口的末尾时）与以下中的任何一个或两个结合时，一些最有用的复合触发器触发一次：

1.  在通过窗口末端的水印之前的推测性点火，以允许更快地处理部分结果。
2.  在水印通过窗口结束后发生的延迟发射，以允许处理迟到的数据

您可以使用AfterWatermark表达此模式。 例如，以下示例触发器代码在以下条件下触发：

1.  在Beam估计所有数据都已到达（水印通过窗口的末尾）
2.  在延迟十分钟后，任何时候迟到的数据都会到来

Other composite triggers

您还可以构建其他种类的复合触发器。 以下示例代码显示了一个简单的复合触发器，只要窗格至少有100个元素，或者在一分钟之后触发。

--------------------

参考：[https://beam.apache.org/documentation/programming-guide][https_beam.apache.org_documentation_programming-guide]

**End**

[WordCount example pipeline]: https://beam.apache.org/get-started/wordcount-example
[Adding Timestamps]: https://beam.apache.org/documentation/programming-guide/#adding-timestamps-to-a-pcollections-elements
[20190214205754772.png]: /images/20220319/f1eac8e58f264317a35cd6ae68f79fe4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70]: /images/20220319/18ff89dc2ea54bcbbaade3c82c65568e.png
[Requirements for writing user code for Beam transforms]: https://beam.apache.org/documentation/programming-guide/#requirements-for-writing-user-code-for-beam-transforms
[Design Your Pipeline]: https://beam.apache.org/documentation/pipelines/design-your-pipeline/#multiple-sources
[non-global windowing]: https://beam.apache.org/documentation/programming-guide/#setting-your-pcollections-windowing-function
[aggregation trigger]: https://beam.apache.org/documentation/programming-guide/#triggers
[GroupByKey and unbounded PCollections]: https://beam.apache.org/documentation/programming-guide/#groupbykey-and-unbounded-pcollections
[Pre-written Beam transforms for Java]: https://beam.apache.org/releases/javadoc/2.10.0/index.html?org/apache/beam/sdk/transforms/package-summary.html
[Pre-written Beam transforms for Python]: https://beam.apache.org/releases/pydoc/2.10.0/apache_beam.transforms.html
[WordCount example program]: https://beam.apache.org/get-started/wordcount-example/
[PTransform Style Guide]: https://beam.apache.org/contribute/ptransform-style-guide/
[number of common data storage types]: https://beam.apache.org/documentation/io/built-in/
[implement your own read and write transforms]: https://beam.apache.org/documentation/io/developing-io-overview/
[20190215213923612.png]: /images/20220319/afa5eca6d7da4940b000f234372e3468.png
[20190219195341174.png]: /images/20220319/8661df2e6caa44a6bbc14fb120024d0e.png
[20190219195725490.png]: /images/20220319/1ebf808616e247f1b28e5869b3df61a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 1]: /images/20220319/a4ff1a13418c41f88891209f915b0921.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 2]: /images/20220319/deeb47d5d4f645af9fdc5a136f3a765b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 3]: /images/20220319/f03809861ea8434885e55556d89797eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0Nzc3NjAw_size_16_color_FFFFFF_t_70 4]: /images/20220319/c15fbe64765b45d0afb68a4e573bbc8e.png
[https_beam.apache.org_documentation_programming-guide]: https://beam.apache.org/documentation/programming-guide/#overview