不多说，直接上干货！

https://beam.apache.org/get-started/beam-overview/

https://beam.apache.org/get-started/quickstart-java/

Apache Beam Java SDK Quickstart

This Quickstart will walk you through executing your first Beam pipeline to run WordCount, written using Beam’s Java SDK, on a runner of your choice.

我这里为了方便大家快速入手，翻译并整理为中文。

本博文通过使用 Java SDK 来完成，你可以尝试运行在不同的执行引擎上。

第一步：设置开发环境

下载并安装 Java Development Kit (JDK) 1.7 或更高版本。检查 JAVA_HOME 环境变量已经设置并指向你的 JDK 安装目录。

照着 Maven 的 安装指南 下载并安装适合你的操作系统的 Apache Maven 。

第二步：获取 示例的WordCount 代码

获得一份 WordCount 管线代码拷贝最简单的方法，就是使用下列指令来生成一个简单的、包含基于 Beam 最新版的 WordCount 示例和构建的 Maven 项目：

Apache Beam 的源代码在 Github 有托管，可以到 Github 下载对应的源码，下载地址：https://github.com/apache/beam

然后，将其中的示例代码进行打包，命令如下所示：(这是最新稳定版本)(所以一般用这个)

$ mvn archetype:generate \

-DarchetypeRepository=https://repository.apache.org/content/groups/snapshots \

-DarchetypeGroupId=org.apache.beam \

-DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \

-DarchetypeVersion=LATEST\

-DgroupId=org.example \

-DartifactId=word-count-beam \

-Dversion="0.1" \

-Dpackage=org.apache.beam.examples \

-DinteractiveMode=false

这是官网推荐的

$ mvn archetype:generate \

-DarchetypeGroupId=org.apache.beam \

-DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \

-DarchetypeVersion=2.1.0 \

-DgroupId=org.example \

-DartifactId=word-count-beam \

-Dversion="0.1" \

-Dpackage=org.apache.beam.examples \

-DinteractiveMode=false

那是因为，最新的Bean为2.1.0。

这将创建一个叫 word-count-beam 的目录，其中包含了一份简单的 pom.xml 文件和一套示例管线，用来计算某个文本文件中的各个单词的数量。

$ cd word-count-beam/

$ ls

pom.xml src

$ ls src/main/java/org/apache/beam/examples/

DebuggingWordCount.java WindowedWordCount.java common

MinimalWordCount.java WordCount.java

关于这些示例中用到的 Beam 的概念的详细介绍，请阅读 WordCount Example Walkthrough 一文。这里我们只聚焦于如何执行 WordCount.java 上。

运行 WordCount 示例代码

一个 Beam 程序可以运行在多个 Beam 的可执行引擎上，包括 ApexRunner，FlinkRunner，SparkRunner 或者 DataflowRunner。 另外还有 DirectRunner。不需要特殊的配置就可以在本地执行，方便测试使用。

下面，你可以按需选择你想执行程序的引擎，即哪个runner后：

对引擎进行相关配置，确保你已经正确配置了该runner。

使用不同的命令：通过 --runner=参数指明引擎类型，默认是 DirectRunner；添加引擎相关的参数；指定输出文件和输出目录，当然这里需要保证文件目录是执行引擎可以访问到的，比如本地文件目录是不能被外部集群访问的。

运行示例程序，你的第一个WordCount 管线。

Direct

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--inputFile=pom.xml --output=counts" -Pdirect-runner

Apex

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--inputFile=pom.xml --output=counts --runner=ApexRunner" -Papex-runner

Flink-Local

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--runner=FlinkRunner --inputFile=pom.xml --output=counts" -Pflink-runner

Flink-Cluster

$ mvn package exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--runner=FlinkRunner --flinkMaster= --filesToStage=target/word-count-beam-bundled-0.1.jar \

--inputFile=/path/to/quickstart/pom.xml --output=/tmp/counts" -Pflink-runner

You can monitor the running job by visiting the Flink dashboard at http://:8081

然后，你可以通过访问 http://:8081 来监测运行的应用程序。

Spark

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--runner=SparkRunner --inputFile=pom.xml --output=counts" -Pspark-runner

Dataflow

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \

-Dexec.args="--runner=DataflowRunner --project= \

--gcpTempLocation=gs:///tmp \

--inputFile=gs://apache-beam-samples/shakespeare/* --output=gs:///counts" \

-Pdataflow-runner

运行结果

当程序运行完成后，你可以看到有多个文件以 count 开头，个数取决于执行引擎的类型。当你查看文件的内容的时候，每个唯一的单词后面会显示其出现次数，但是前后顺序是不固定的，也是分布式引擎为了提高效率的一种常用方式。

一旦管线完成运行，你可以查看结果。你会注意到有多个以 count 打头的输出文件。具体会有几个这样的文件是由 runner 决定的。这样能方便 runner 进行高效的分布式执行。

当你查看文件内容的时候，你会看到里面包含每个单词的出现数量。文件中的元素顺序可能会和这里看到的不同。因为 Beam 模型通常并不保障顺序，以便于 runner 优化效率。

Direct

$ ls counts*

$ more counts*

api: 9

bundled: 1

old: 4

Apache: 2

The: 1

limitations: 1

Foundation: 1

...

Apex

$ cat counts*

BEAM: 1

have: 1

simple: 1

skip: 4

PAssert: 1

...

Flink-Local

$ ls counts*

$ more counts*

The: 1

api: 9

old: 4

Apache: 2

limitations: 1

bundled: 1

Foundation: 1

...

Flink-Cluster

$ ls /tmp/counts*

$ more /tmp/counts*

The: 1

api: 9

old: 4

Apache: 2

limitations: 1

bundled: 1

Foundation: 1

...

Spark

$ ls counts*

$ more counts*

beam: 27

SF: 1

fat: 1

job: 1

limitations: 1

require: 1

of: 11

profile: 10

...

Dataflow

$ gsutil ls gs:///counts*

$ gsutil cat gs:///counts*

feature: 15

smother'st: 1

revelry: 1

bashfulness: 1

Bashful: 1

Below: 2

deserves: 32

barrenly: 1

...

总结

Apache Beam 主要针对理想并行的数据处理任务，并通过把数据集拆分多个子数据集，让每个子数据集能够被单独处理，从而实现整体数据集的并行化处理。当然，也可以用 Beam 来处理抽取，转换和加载任务和数据集成任务(一个ETL过程)。进一步将数据从不同的存储介质中或者数据源中读取，转换数据格式，最后加载到新的系统中。