Apache Beam实战指南 | 玩转KafkaIO与Flink

AI前线导读：本文是 **Apache Beam实战指南系列文章** 的第二篇内容，将重点介绍 Apache Beam与Flink的关系，对Beam框架中的KafkaIO和Flink源码进行剖析，并结合应用示例和代码解读带你进一步了解如何结合Beam玩转Kafka和Flink。系列文章第一篇回顾 Apache Beam实战指南之基础入门

一．概述

大数据发展趋势从普通的大数据，发展成AI大数据，再到下一代号称万亿市场的lOT大数据。技术也随着时代的变化而变化，从Hadoop的批处理，到Spark Streaming，以及流批处理的Flink的出现，整个大数据架构也在逐渐演化。

Apache Beam作为新生技术，在这个时代会扮演什么样的角色，跟Flink之间的关系是怎样的？Apache Beam和Flink的结合会给大数据开发者或架构师们带来哪些意想不到的惊喜呢？

二．大数据架构发展演进历程

2.1 大数据架构Hadoop

图2-1 MapReduce 流程图

最初做大数据是把一些日志或者其他信息收集后写入Hadoop 的HDFS系统中，如果运营人员需要报表，则利用Hadoop的MapReduce进行计算并输出，对于一些非计算机专业的统计人员，后期可以用Hive进行统计输出。

2.2 流式处理Storm

图2-2Storm流程图

业务进一步发展，运营人员需要看到实时数据的展示或统计。例如电商网站促销的时候，用于统计用户实时交易数据。数据收集也使用MQ，用流式Storm解决这一业务需求问题。

2.3 Spark批处理和微批处理

图2-3 Spark流程图

业务进一步发展，服务前端加上了网关进行负载均衡，消息中心也换成了高吞吐量的轻量级MQ Kafka，数据处理渐渐从批处理发展到微批处理。

2.4 Flink：真正的流批处理统一

图2-4 Flink 流程图

随着AI和loT的发展，对于传感设备的信息、报警器的警情以及视频流的数据量微批计算引擎已经满足不了业务的需求，Flink实现真正的流处理让警情更实时。

2.5 下一代大数据处理统一标准Apache Beam

图2-5      Apache Beam 流程图

BeamSDKs封装了很多的组件IO，也就是图左边这些重写的高级API，使不同的数据源的数据流向后面的计算平台。通过将近一年的发展，Apache Beam 不光组件IO更加丰富了，并且计算平台在当初最基本的 Apache Apex、Direct Runner、Apache Flink、Apache Spark、Google Cloud Dataflow之上，又增加了Gearpump、Samza 以及第三方的JStorm等计算平台。

为什么说Apache Beam 会是大数据处理统一标准呢？

因为很多现在大型公司都在建立自己的“大中台”，建立统一的数据资源池，打通各个部门以及子公司的数据，以解决信息孤岛问题，把这些数据进行集中式管理并且进行后期的数据分析、BI、AI以及机器学习等工作。这种情况下会出现很多数据源，例如之前用的MySQL、MongodDB、HDFS、HBase、Solr 等，如果想建立中台就会是一件令人非常苦恼的事情，并且多计算环境更是让技术领导头疼。Apache Beam的出现正好迎合了这个时代的新需求，它集成了很多数据库常用的数据源并把它们封装成SDK的IO，开发人员没必要深入学习很多技术，只要会写Beam 程序就可以了，大大节省了人力、时间以及成本。

三．Apache Beam和Flink的关系

随着阿里巴巴Blink的开源，Flink中国社区开始活跃起来。很多人会开始对各种计算平台进行对比，比如Storm、Spark、JStorm、Flink等，并且有人提到之前阿里巴巴开源的JStorm比Flink性能高出10-15倍，为什么阿里巴巴却转战基于Flink的Blink呢? 在最近Flink的线下技术会议上，阿里巴巴的人已经回答了这一问题。其实很多技术都是从业务实战出来的，随着业务的发展可能还会有更多的计算平台出现，没有必要对此过多纠结。

不过，既然大家最近讨论得这么火热，这里也列出一些最近问的比较多的、有代表性的关于Beam的问题，逐一进行回答。

1. Flink支持SQL，请问Beam支持吗？

现在Beam是支持SQL处理的，底层技术跟Flink底层处理是一样的。

Beam SQL现在只支持Java，底层是Apache Calcite 的一个动态数据管理框架，用于大数据处理和一些流增强功能，它允许你自定义数据库功能。例如Hive 使用了Calcite的查询优化，当然还有Flink解析和流SQL处理。Beam在这之上添加了额外的扩展，以便轻松利用Beam的统一批处理/流模型以及对复杂数据类型的支持。 以下是Beam SQL具体处理流程图：

Beam SQL一共有两个比较重要的概念：

SqlTransform：用于PTransforms从SQL查询创建的接口。

Row：Beam SQL操作的元素类型。例如：PCollection<Row>。

在将SQL查询应用于PCollection 之前，集合中Row的数据格式必须要提前指定。 一旦Beam SQL 指定了 管道中的类型是不能再改变的。PCollection行中字段/列的名称和类型由Schema进行关联定义。您可以使用Schema.builder()来创建 Schemas。

示例：

// Define the schema for the records.
Schema appSchema = 
    Schema
      .builder()
      .addInt32Field("appId")
      .addStringField("description")
      .addDateTimeField("rowtime")
      .build();
// Create a concrete row with that type.
Row row = 
    Row
      .withSchema(appSchema)
      .addValues(1, "Some cool app", new Date())
      .build();
// Create a source PCollection containing only that row
 PCollection<Row> testApps = 
    PBegin
      .in(p)
      .apply(Create
                .of(row)
                .withCoder(appSchema.getRowCoder()));

也可以是其他类型，不是直接是Row，利用 PCollection<T> 通过应用ParDo可以将输入记录转换为Row格式。如：

// An example POJO class.
 class AppPojo {
  Integer appId;
  String description;
  Date timestamp;
}
// Acquire a collection of POJOs somehow.
  PCollection<AppPojo> pojos = ...
// Convert them to Rows with the same schema as defined above via a DoFn.
PCollection<Row> apps = pojos
  .apply(
      ParDo.of(new DoFn<AppPojo, Row>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
          // Get the current POJO instance
          AppPojo pojo = c.element();
          // Create a Row with the appSchema schema 
            // and values from the current POJO
            Row appRow = 
                  Row
                    .withSchema(appSchema)
                   .addValues(
                      pojo.appId, 
                      pojo.description, 
                     pojo.timestamp)
                    .build();
          // Output the Row representing the current POJO
          c.output(appRow);
        }
      }));

2. Flink 有并行处理，Beam 有吗？

Beam 在抽象Flink的时候已经把这个参数抽象出来了，在Beam Flink 源码解析中会提到。

3. 我这里有个流批混合的场景，请问Beam是不是支持？

这个是支持的，因为批也是一种流，是一种有界的流。Beam 结合了Flink，Flink dataset 底层也是转换成流进行处理的。

4. Flink流批写程序的时候和Beam有什么不同？底层是Flink还是Beam?

打个比喻，如果Flink是Lucene，那么Beam 就是Solr，把Flink 的API进行二次重写，简化了API，让大家使用更简单、更方便。此外，Beam提供了更多的数据源，这是Flink不能比的。当然，Flink 后期可能也会往这方面发展。

四．Apache Beam KafkaIO源码剖析

Apache Beam KafkaIO 对kafka-clients支持依赖情况

KafkaIO是Kafka的API封装，主要负责Apache Kafka读取和写入消息。如果想使用KafkaIO，必须依赖beam-sdks-java-io-kafka ，KafkaIO 同时支持多个版本的Kafka客户端，使用时建议用高版本的或最新的Kafka 版本，因为使用KafkaIO的时候需要包含kafka-clients 的依赖版本。

Apache Beam KafkaIO 对各个kafka-clients 版本的支持情况如下表：

表4-1 KafkaIO 与kafka-clients 依赖关系表

Apache Beam V2.1.0版本之前源码中的pom文件都显式指定了特定的0.9.0.1版本支持，但是从V2.1.0版本和V2.1.1两个版本开始已经替换成了kafka-clients 的0.10.1.0 版本，并且源码中提示0.10.1.0 版本更安全。这是因为去年Kafka 0.10.1.0 之前的版本曝出了安全漏洞。在V2.2.0 以后的版本中，Beam对API做了调整和更新，对之前的两种版本都支持，不过需要在pom中引用的时候自己指定Kafka的版本。但是在Beam V2.5.0 和V2.6.0 版本，源码中添加了以下提示：

* <h3>Supported Kafka Client Versions</h3>
 * KafkaIO relies on <i>kafka-clients</i> for all its interactions with the Kafka cluster.
 * <i>kafka-clients</i> versions 0.10.1 and newer are supported at runtime. The older versions
 * 0.9.x - 0.10.0.0 are also supported, but are deprecated and likely be removed in near future.
 * Please ensure that the version included with the application is compatible with the version of
 * your Kafka cluster. Kafka client usually fails to initialize with a clear error message in
 * case of incompatibility.
 */

也就说在这两个版本已经移除了对Kafka 客户端 0.10.1.0 以前版本的支持，旧版本还会支持，但是在以后不久就会删除。所以大家在使用的时候要注意版本的依赖关系和客户端的版本支持度。

如果想使用KafkaIO，pom 必须要引用，版本跟4-1表中的对应起来就可以了。

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-kafka</artifactId>
    <version>...</version>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>a_recent_version</version>
  <scope>runtime</scope>
</dependency>

KafkaIO读写源码解析

KafkaIO源码链接如下：

链接

在KafkaIO里面最主要的两个方法是Kafka的读写方法。

KafkaIO读操作

pipeline.apply(KafkaIO.<Long, String>read() 
        .withBootstrapServers("broker_1:9092,broker_2:9092")// 
        .withTopic("my_topic")    // use withTopics(List<String>) to read from multiple topics. 
        .withKeyDeserializer(LongDeserializer.class) 
       .withValueDeserializer(StringDeserializer.class) 
       // Above four are required configuration. returns   PCollection<KafkaRecord<Long, String>> 
       // Rest of the settings are optional :  
        // you can further customize KafkaConsumer used to read the records by   adding more 
       // settings for ConsumerConfig. e.g : 
       .updateConsumerProperties(ImmutableMap.of("group.id",   "my_beam_app_1")) 
 
        // set event times and watermark based on 'LogAppendTime'. To provide   a custom 
       // policy see withTimestampPolicyFactory(). withProcessingTime() is   the default. 
       // Use withCreateTime() with topics that have 'CreateTime' timestamps. 
        .withLogAppendTime()  
       // restrict reader to committed messages on Kafka (see method   documentation). 
        .withReadCommitted()  
        // offset consumed by the pipeline can be committed back. 
        .commitOffsetsInFinalize()  
        // finally, if you don't need Kafka metadata, you can drop it.g 
       .withoutMetadata() // PCollection<KV<Long, String>> 
     ) 
     .apply(Values.<String>create()) // PCollection<String>

1) 指定KafkaIO的模型，从源码中不难看出这个地方的KafkaIO<K,V>类型是Long和String 类型，也可以换成其他类型。

pipeline.apply(KafkaIO.<Long, String>read() pipeline.apply(KafkaIO.<Long, String>read()

2) 设置Kafka集群的集群地址。

.withBootstrapServers("broker_1:9092,broker_2:9092")

3) 设置Kafka的主题类型，源码中使用了单个主题类型，如果是多个主题类型则用withTopics(List<String>)方法进行设置。设置情况基本跟Kafka原生是一样的。

.withTopic("my_topic") // use withTopics(List<String>) to read from multiple topics.

4) 设置序列化类型。Apache Beam KafkaIO 在序列化的时候做了很大的简化，例如原生Kafka可能要通过Properties 类去设置 ，还要加上很长一段jar包的名字。

Beam KafkaIO的写法：

.withKeyDeserializer(LongDeserializer.class) 
 .withValueDeserializer(StringDeserializer.class)

原生Kafka的设置：

Properties props = new Properties();
props.put("key.deserializer","org.apache.kafka.common.serialization.ByteArrayDeserializer");
props.put("value.deserializer","org.apache.kafka.common.serialization.ByteArrayDeserializer");

5) 设置Kafka的消费者属性，这个地方还可以设置其他的属性。源码中是针对消费分组进行设置。

.updateConsumerProperties(ImmutableMap.of("group.id", my_beam_app_1"))

6) 设置Kafka吞吐量的时间戳，可以是默认的，也可以自定义。

.withLogAppendTime()

7) 相当于Kafka 中"isolation.level", "read_committed" ，指定KafkaConsumer只应读取非事务性消息，或从其输入主题中提交事务性消息。流处理应用程序通常在多个读取处理写入阶段处理其数据，每个阶段使用前一阶段的输出作为其输入。通过指定read_committed模式，我们可以在所有阶段完成一次处理。针对"Exactly-once" 语义，支持Kafka 0.11版本。

.withReadCommitted()

8) 设置Kafka是否自动提交属性"AUTO_COMMIT"，默认为自动提交，使用Beam 的方法来设置。

set CommitOffsetsInFinalizeEnabled(boolean commitOffsetInFinalize)
.commitOffsetsInFinalize()

9) 设置是否返回Kafka的其他数据，例如offset 信息和分区信息，不用可以去掉。

.withoutMetadata() // PCollection<KV<Long, String>>

10) 设置只返回values值，不用返回key。例如 PCollection<String>，而不是PCollection<Long，String>。

.apply(Values.<String>create()) // PCollection<String>

KafkaIO写操作

写操作跟读操作配置基本相似，我们看一下具体代码。

PCollection<KV<Long, String>> kvColl = ...;
kvColl.apply(KafkaIO.<Long, String>write()
      .withBootstrapServers("broker_1:9092,broker_2:9092")
      .withTopic("results")
      .withKeySerializer(LongSerializer.class)
      .withValueSerializer(StringSerializer.class)
      // You can further customize KafkaProducer used to write the records by adding more
      // settings for ProducerConfig. e.g, to enable compression :
      .updateProducerProperties(ImmutableMap.of("compression.type", "gzip"))
     // You set publish timestamp for the Kafka records.
      .withInputTimestamp() // element timestamp is used while publishing to Kafka
      // or you can also set a custom timestamp with a function.
      .withPublishTimestampFunction((elem, elemTs) -> ...)
      // Optionally enable exactly-once sink (on supported runners). See JavaDoc for withEOS().
      .withEOS(20, "eos-sink-group-id");
   );

下面这个是Kafka里面比较重要的一个属性设置，在Beam中是这样使用的，非常简单，但是要注意这个属性.withEOS 其实就是Kafka中"Exactly-once"。

.withEOS(20, "eos-sink-group-id");

在写入Kafka时完全一次性地提供语义，这使得应用程序能够在Beam管道中的一次性语义之上提供端到端的一次性保证。它确保写入接收器的记录仅在Kafka上提交一次，即使在管道执行期间重试某些处理也是如此。重试通常在应用程序重新启动时发生（如在故障恢复中）或者在重新分配任务时（如在自动缩放事件中）。Flink runner通常为流水线的结果提供精确一次的语义，但不提供变换中用户代码的副作用。如果诸如Kafka接收器之类的转换写入外部系统，则这些写入可能会多次发生。

在此处启用EOS时，接收器转换将兼容的Beam Runners中的检查点语义与Kafka中的事务联系起来，以确保只写入一次记录。由于实现依赖于runners checkpoint语义，因此并非所有runners都兼容。Beam中FlinkRunner针对Kafka 0.11+版本才支持，然而Dataflow runner和Spark runner如果操作kafkaIO是完全支持的。

关于性能的注意事项

"Exactly-once" 在接收初始消息的时候，除了将原来的数据进行格式化转换外，还经历了2个序列化 - 反序列化循环。根据序列化的数量和成本，CPU可能会涨的很明显。通过写入二进制格式数据（即在写入Kafka接收器之前将数据序列化为二进制数据）可以降低CPU成本。

关于参数

numShards——设置接收器并行度。存储在Kafka上的状态元数据，使用sinkGroupId存储在许多虚拟分区中。一个好的经验法则是将其设置为Kafka主题中的分区数。

sinkGroupId——用于在Kafka上将少量状态存储为元数据的组ID。它类似于与KafkaConsumer一起使用的使用groupID。每个作业都应使用唯一的groupID，以便重新启动/更新作业保留状态以确保一次性语义。状态是通过Kafka上的接收器事务原子提交的。有关更多信息，请参阅KafkaProducer.sendOffsetsToTransaction（Map，String）。接收器在初始化期间执行多个健全性检查以捕获常见错误，以便它不会最终使用似乎不是由同一作业写入的状态。

五．Apache Beam Flink源码剖析

Apache Beam FlinkRunner对 Flink支持依赖情况

Flink 是一个流和批处理的统一的计算框架，Apache Beam 跟Flink API做了无缝集成。在Apache Beam中对Flink 的操作主要是 FlinkRunner.java，Apache Beam支持不同版本的flink 客户端。我根据不同版本列了一个Flink 对应客户端支持表如下：

图5-1 FlinkRunner与Flink依赖关系表

从图5-1中可以看出，Apache Beam 对Flink 的API支持的更新速度非常快，从源码可以看到2.0.0版本之前的FlinkRunner是非常low的，并且直接拿Flink的实例做为Beam的实例，封装的效果也比较差。但是从2.0.0 版本之后 ，Beam就像打了鸡血一样API更新速度特别快，抛弃了以前的冗余，更好地跟Flink集成，让人眼前一亮。

Apache Beam Flink 源码解析

因为Beam在运行的时候都是显式指定Runner，在FlinkRunner源码中只是成了简单的统一入口，代码非常简单，但是这个入口中有一个比较关键的接口类FlinkPipelineOptions。

请看代码：

/** Provided options. */
private final FlinkPipelineOptions options;

通过这个类我们看一下Apache Beam到底封装了哪些Flink方法。

首先FlinkPipelineOptions是一个接口类，但是它继承了 PipelineOptions、ApplicationNameOptions、StreamingOptions 三个接口类，第一个PipelineOptions大家应该很熟悉了，用于基本管道创建；第二个ApplicationNameOptions 用于设置应用程序名字；第三个用于判断是流式数据还是批数据。源代码如下：

public interface FlinkPipelineOptions  extends 
PipelineOptions, ApplicationNameOptions, StreamingOptions {
  //....
}

1) 设置 Flink Master 方法 ，这个方法用于设置Flink 集群地址的Master地址。可以填写IP和端口，或者是hostname 和端口，默认local 。当然测试也可以是单机的，在Flink 1.4 利用 start-local.sh 启动，而到了1.5以上就去掉了这个脚本，本地直接换成了 start-cluster.sh。大家测试的时候需要注意一下。

/**
* The url of the Flink JobManager on which to execute pipelines.   This can either be the the   * address of a cluster JobManager, in the form "host:port" or one of the special Strings  * "[collection]" will execute the pipeline on Java Collections while "[auto]" will let the system
 */
@Description( "Address of the Flink Master where the Pipeline should  be executed. Can"+ "[collection] or [auto].")
void setFlinkMaster(String value);

2) 设置 Flink 的并行数，属于Flink 高级API里面的属性。设置合适的parallelism能提高运算效率，太多了和太少了都不行。设置parallelism有多种方式，优先级为api>env>p>file。

@Description("The degree of parallelism to be used when distributing operations onto workers.")
@Default.InstanceFactory(DefaultParallelismFactory.class)
Integer getParallelism();
void setParallelism(Integer value);

3) 设置连续检查点之间的间隔时间（即当前的快照）用于容错的管道状态。

@Description("The interval between consecutive checkpoints (i.e.  snapshots of the current"
@Default.Long(-1L)
Long getCheckpointingInterval();
void setCheckpointingInterval(Long interval)

4) 定义一致性保证的检查点模式，默认为"AT_LEAST_ONCE"，在Beam的源码中定义了一个枚举类CheckpointingMode，除了默认的"AT_LEAST_ONCE"，还有"EXACTLY_ONCE"。

"AT_LEAST_ONCE"：这个模式意思是系统将以一种更简单地方式来对operator和udf的状态进行快照：在失败后进行恢复时，在operator的状态中，一些记录可能会被重放多次。

"EXACTLY_ONCE"：这种模式意思是系统将以如下语义对operator和udf(user defined function)进行快照：在恢复时，每条记录将在operator状态中只被重现/重放一次。

@Description("The checkpointing mode that defines consistency guarantee.")
@Default.Enum("AT_LEAST_ONCE")
CheckpointingMode getCheckpointingMode();
void setCheckpointingMode(CheckpointingMode mode);

5) 设置检查点的最大超时时间，默认为20*60*1000(毫秒)=20(分钟)。

@Description("The maximum time that a checkpoint may take before being discarded.")
@Default.Long(20 * 60 * 1000)
Long getCheckpointTimeoutMillis();
void setCheckpointTimeoutMillis(Long checkpointTimeoutMillis);

6) 设置重新执行失败任务的次数，值为0有效地禁用容错，值为-1表示使用系统默认值（在配置中定义)。

@Description(
"Sets the number of times that failed tasks are re-executed. "
+ "A value of zero effectively disables fault tolerance. A value of -1 indicates "+ "that the system default value (as defined in the configuration) should be used.")
@Default.Integer(-1)
Integer getNumberOfExecutionRetries();
void setNumberOfExecutionRetries(Integer retries);

7) 设置执行之间的延迟，默认值为-1L。

@Description(
      "Sets the delay between executions. A value of {@code -1} "
          + "indicates that the default value should be used.")
@Default.Long(-1L)
Long getExecutionRetryDelay();
void setExecutionRetryDelay(Long delay);

8) 设置重用对象的行为。

@Description("Sets the behavior of reusing objects.")
@Default.Boolean(false)
Boolean getObjectReuse();
void setObjectReuse(Boolean reuse);

9) 设置状态后端在计算期间存储Beam的状态，不设置从配置文件中读取默认值。注意：仅在执行时适用流媒体模式。

@Description("Sets the state backend to use in streaming mode. "
@JsonIgnore
AbstractStateBackend getStateBackend();
void setStateBackend(AbstractStateBackend stateBackend);

10) 在Flink Runner中启用/禁用Beam指标。

@Description("Enable/disable Beam metrics in Flink Runner")
@Default.Boolean(true)
BooleangetEnableMetrics();
voidsetEnableMetrics(BooleanenableMetrics);

11) 启用或禁用外部检查点，与CheckpointingInterval一起使用。

@Description(
"Enables or disables externalized checkpoints."
+"Works in conjunction with CheckpointingInterval")
@Default.Boolean(false)
BooleanisExternalizedCheckpointsEnabled();
voidsetExternalizedCheckpointsEnabled(BooleanexternalCheckpoints);

12) 设置当他们的Wartermark达到+ Inf时关闭源，Watermark在Flink 中其中一个作用是根据时间戳做单节点排序，Beam也是支持的。

@Description("If set, shutdown sources when their watermark reaches +Inf.")
@Default.Boolean(false)
BooleanisShutdownSourcesOnFinalWatermark();
voidsetShutdownSourcesOnFinalWatermark(BooleanshutdownOnFinalWatermark);

剩余两个部分这里不再进行翻译，留给大家去看源码。

六. KafkaIO和Flink实战

本节通过解读一个真正的KafkaIO和Flink实战案例，帮助大家更深入地了解Apache Beam KafkaIO和Flink的运用。

设计架构图和设计思路解读

Apache Beam 外部数据流程图

设计思路：Kafka消息生产程序发送testmsg到Kafka集群，Apache Beam 程序读取Kafka的消息，经过简单的业务逻辑，最后发送到Kafka集群，然后Kafka消费端消费消息。

Apache Beam 内部数据处理流程图

Apache Beam 程序通过kafkaIO读取Kafka集群的数据，进行数据格式转换。数据统计后，通过KafkaIO写操作把消息写入Kafka集群。最后把程序运行在Flink的计算平台上。

软件环境和版本说明

• 系统版本 centos 7
• Kafka集群版本： kafka_2.10-0.10.1.1.tgz
• Flink 版本：flink-1.5.2-bin-hadoop27-scala_2.11.tgz

Kafka集群和Flink单机或集群配置，大家可以去网上搜一下配置文章，操作比较简单，这里就不赘述了。

实践步骤

1）新建一个Maven项目

2）在pom文件中添加jar引用

<dependency>
   <groupId>org.apache.beam</groupId>
   <artifactId>beam-sdks-java-io-kafka</artifactId>
   <version>2.4.0</version>
  </dependency>
  <dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>0.10.1.1</version>
  </dependency>
  <dependency>
   <groupId>org.apache.beam</groupId>
   <artifactId>beam-runners-core-java</artifactId>
   <version>2.4.0</version>
  </dependency>
  <dependency>
   <groupId>org.apache.beam</groupId>
   <artifactId>beam-runners-flink_2.11</artifactId>
   <version>2.4.0</version>
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-java</artifactId>
   <version>1.5.2</version>
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-clients_2.11</artifactId>
   <version>1.5.2</version>
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-core</artifactId>
  <version>1.5.2</version>
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-runtime_2.11</artifactId>
   <version>1.5.2</version>
   <!--<scope>provided</scope>-->
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-streaming-java_2.11</artifactId>
   <version>1.5.2</version>
   <!--<scope>provided</scope>-->
  </dependency>
  <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-metrics-core</artifactId>
   <version>1.5.2</version>
   <!--<scope>provided</scope>-->
  </dependency>

3）新建BeamFlinkKafka.java类

4）编写以下代码：

public static void main(String[] args) {
//创建管道工厂
PipelineOptions options = PipelineOptionsFactory.create(); 
// 显式指定PipelineRunner：FlinkRunner必须指定如果不制定则为本地 
 options.setRunner(FlinkRunner.class); 
//设置相关管道
 Pipeline pipeline = Pipeline.create(options);
 //这里kV后说明kafka中的key和value均为String类型
PCollection<KafkaRecord<String, String>> lines = 
pipeline.apply(KafkaIO.<String, 
// 必需设置kafka的服务器地址和端口
String>read().withBootstrapServers("192.168.1.110:11092,192.168.1.119:11092,192.168.1.120:11092")
      .withTopic("testmsg")// 必需设置要读取的kafka的topic名称
      .withKeyDeserializer(StringDeserializer.class)// 必需序列化key
      .withValueDeserializer(StringDeserializer.class)// 必需序列化value
      .updateConsumerProperties(ImmutableMap.<String, Object>of("auto.offset.reset", "earliest")));//这个属性kafka最常见的.
  // 为输出的消息类型。或者进行处理后返回的消息类型
PCollection<String> kafkadata = lines.apply("Remove Kafka Metadata", ParDo.of(new DoFn<KafkaRecord<String, String>, String>() { 
 private static final long serialVersionUID = 1L;
   @ProcessElement
   public void processElement(ProcessContext ctx) {
    System.out.print("输出的分区为----：" + ctx.element().getKV());
    ctx.output(ctx.element().getKV().getValue());// 其实我们这里是把"张海     涛在发送消息***"进行返回操作
   }
  }));
PCollection<String> windowedEvents = kafkadata.apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(5))));
 PCollection<KV<String, Long>> wordcount = windowedEvents.apply(Count.<String>perElement()); // 统计每一个kafka消息的Count
PCollection<String> wordtj = wordcount.apply("ConcatResultKVs", MapElements.via( // 拼接最后的格式化输出（Key为Word，Value为Count）
  new SimpleFunction<KV<String, Long>, String>() {
   private static final long serialVersionUID = 1L;
    @Override
    public String apply(KV<String, Long> input) {
    System.out.print("进行统计：" + input.getKey() + ": " + input.getValue());
      return input.getKey() + ": " + input.getValue();
     }
    }));
  wordtj.apply(KafkaIO.<Void, String>write()  .withBootstrapServers("192.168.1.110:11092,192.168.1.119:11092,192.168.1.120:11092")//设置写会kafka的集群配置地址
    .withTopic("senkafkamsg")//设置返回kafka的消息主题
    // .withKeySerializer(StringSerializer.class)//这里不用设置了，因为上面 Void 
    .withValueSerializer(StringSerializer.class)
    // Dataflow runner and Spark 兼容， Flink 对kafka0.11才支持。我的版本是0.10不兼容
    //.withEOS(20, "eos-sink-group-id")
    .values() // 只需要在此写入默认的key就行了，默认为null值
  ); // 输出结果
  pipeline.run().waitUntilFinish();
｝

5）打包jar，本示例是简单的实战，并没有用Docker，Apache Beam新版本是支持Docker的。

6）通过Apache Flink Dashboard 提交job

7）查看结果

程序接收的日志如下：

七．实战解析

本次实战在源码分析中已经做过详细解析，在这里不做过多的描述，只选择部分问题再重点解释一下。此外，如果还没有入门，甚至连管道和Runner等概念都还不清楚，建议先阅读本系列的第一篇文章 《Apache Beam实战指南之基础入门》 。

1.FlinkRunner在实战中是显式指定的，如果想设置参数怎么使用呢？其实还有另外一种写法，例如以下代码：

//FlinkPipelineOptions options =PipelineOptionsFactory.as(FlinkPipelineOptions.class);
 //options.setStreaming(true);
 //options.setAppName("app_test");
 //options.setJobName("flinkjob");
 //options.setFlinkMaster("localhost:6123");
 //options.setParallelism(10);//设置flink 的并行度
//显式指定PipelineRunner：FlinkRunner，必须指定，如果不指定则为本地 
 options.setRunner(FlinkRunner.class);

2.Kafka 有三种数据读取类型，分别是 “earliest ”，“latest ”，“none ”，分别的意思代表是：

earliest

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 。

latest

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 。

none

topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。

.updateConsumerProperties(ImmutableMap.<String,Object>of("auto.offset.reset", "earliest")));

3.实战中我自己想把Kafka的数据写入，key不想写入，所以出现了Kafka的key项为空，而values才是真正发送的数据。所以开始和结尾要设置个.values()，如果不加上就会报错。

KafkaIO.<Void, String>write()
.values() // 只需要在此写入默认的key就行了，默认为null值

八．小结

随着AI和loT的时代的到来，各个公司不同结构、不同类型、不同来源的数据进行整合的成本越来越高。Apache Beam 技术的统一模型和大数据计算平台特性优雅地解决了这一问题，相信在loT万亿市场中，Apache Beam将会发挥越来越重要的角色。

作者介绍

张海涛，目前就职于海康威视云基础平台，负责云计算大数据的基础架构设计和中间件的开发，专注云计算大数据方向。Apache Beam 中文社区发起人之一，如果想进一步了解最新 Apache Beam 动态和技术研究成果，请加微信 cyrjkj 入群共同研究和运用。