Apache Beam，批处理和流式处理的融合！

1. 概述

在本教程中，我们将介绍 Apache Beam 并探讨其基本概念。

我们将首先演示使用 Apache Beam 的用例和好处，然后介绍基本概念和术语。之后，我们将通过一个简单的例子来说明 Apache Beam 的所有重要方面。

2. Apache Beam是个啥?

Apache Beam（Batch+strEAM）是一个用于批处理和流式数据处理作业的统一编程模型。它提供了一个软件开发工具包，用于定义和构建数据处理管道以及执行这些管道的运行程序。

Apache Beam旨在提供一个可移植的编程层。事实上，Beam管道运行程序将数据处理管道转换为与用户选择的后端兼容的API。目前，支持这些分布式处理后端有：

• Apache Apex
• Apache Flink
• Apache Gearpump (incubating)
• Apache Samza
• Apache Spark
• Google Cloud Dataflow
• Hazelcast Jet

3. 为啥选择 Apache Beam

Apache Beam 将批处理和流式数据处理融合在一起，而其他组件通常通过单独的 API 来实现这一点。因此，很容易将流式处理更改为批处理，反之亦然，例如，随着需求的变化。

Apache Beam 提高了可移植性和灵活性。我们关注的是逻辑，而不是底层的细节。此外，我们可以随时更改数据处理后端。

Apache Beam 可以使用 Java、Python、Go和 Scala等SDK。事实上，团队中的每个人都可以使用他们选择的语言。

4. 基本概念

使用 Apache Beam，我们可以构建工作流图（管道）并执行它们。编程模型中的关键概念是：

• PCollection–表示可以是固定批处理或数据流的数据集
• PTransform–一种数据处理操作，它接受一个或多个 PCollections 并输出零个或多个 PCollections。
• Pipeline–表示 PCollection 和 PTransform 的有向无环图，因此封装了整个数据处理作业。
• PipelineRunner–在指定的分布式处理后端上执行管道。

简单地说，PipelineRunner 执行一个管道，管道由 PCollection 和 PTransform 组成。

5. 字数统计示例

现在我们已经学习了 Apache Beam 的基本概念，让我们设计并测试一个单词计数任务。

5.1 建造梁式管道

设计工作流图是每个 Apache Beam 作业的第一步，单词计数任务的步骤定义如下：

1.从原文中读课文。

2.把课文分成单词表。

3.所有单词都小写。

4.删去标点符号。

5.过滤停止语。

6.统计唯一单词数量。

为了实现这一点，我们需要使用 PCollection 和 PTransform 抽象将上述步骤转换为 管道 。

5.2. 依赖

在实现工作流图之前，先添加 Apache Beam的依赖项 到我们的项目：

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
    <version>${beam.version}</version>
</dependency>

Beam管道运行程序依赖于分布式处理后端来执行任务。我们添加 DirectRunner 作为运行时依赖项：

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-direct-java</artifactId>
    <version>${beam.version}</version>
    <scope>runtime</scope>
</dependency>

与其他管道运行程序不同， DirectRunner 不需要任何额外的设置，这对初学者来说是个不错的选择。

5.3. 实现

Apache Beam 使用 Map-Reduce 编程范式 ( 类似 Java Stream)。讲下面内容之前，最好 对 reduce() , filter() , count() , map() , 和 flatMap() 有个基础概念和认识。

首先要做的事情就是 创建 管道 ：

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);

六步单词计数任务：

PCollection<KV<String, Long>> wordCount = p
    .apply("(1) Read all lines", 
      TextIO.read().from(inputFilePath))
    .apply("(2) Flatmap to a list of words", 
      FlatMapElements.into(TypeDescriptors.strings())
      .via(line -> Arrays.asList(line.split("\\s"))))
    .apply("(3) Lowercase all", 
      MapElements.into(TypeDescriptors.strings())
      .via(word -> word.toLowerCase()))
    .apply("(4) Trim punctuations", 
      MapElements.into(TypeDescriptors.strings())
      .via(word -> trim(word)))
    .apply("(5) Filter stopwords", 
      Filter.by(word -> !isStopWord(word)))
    .apply("(6) Count words", 
      Count.perElement());

apply() 的第一个（可选）参数是一个 String ，它只是为了提高代码的可读性。下面是上述代码中每个 apply() 的作用：

1. 首先，我们使用 TextIO 逐行读取输入文本文件。
2. 将每一行按空格分开，把它映射到一个单词表上。
3. 单词计数不区分大小写，所以我们将所有单词都小写。
4. 之前，我们用空格分隔行，但是像“word！“和”word？"这样的，就需要删除标点符号。
5. 像“is”和“by”这样的停止词在几乎每一篇英语文章中都很常见，所以我们将它们删除。
6. 最后，我们使用内置函数 Count.perElement() 计算唯一单词数量。

如前所述，管道是在分布式后端处理的。不可能在内存中的 PCollection 上迭代，因为它分布在多个后端。相反，我们将结果写入外部数据库或文件。

首先，我们将 PCollection 转换为 String 。然后，使用 TextIO 编写输出：

wordCount.apply(MapElements.into(TypeDescriptors.strings())
    .via(count -> count.getKey() + " --> " + count.getValue()))
    .apply(TextIO.write().to(outputFilePath));

现在 管道 已经定义好了，接下来做个简单的测试。

5.4. 运行测试

到目前为止，我们已为单词计数任务定义了 管道 ，现在运行 管道 ：

p.run().waitUntilFinish();

在这行代码中，Apache Beam 将把我们的任务发送到多个 DirectRunner 实例。因此，最后将生成几个输出文件。它们将包含以下内容：

...
apache --> 3
beam --> 5
rocks --> 2
...

在 Apache Beam 中定义和运行分布式作业是如此地简单。为了进行比较，单词计数实现在 Apache Spark, Apache Flink 和 Hazelcast-Jet 上也有

6. 结语