本文是 Apache Beam实战指南系列文章 的第三篇内容，将对Beam框架中的HDFSIO和MySQLIO源码进行剖析，并结合应用示例和代码解读带你进一步了解如何结合Beam玩转大数据存储重要组件HDFS。

系列文章第一篇回顾：Apache Beam实战指南 | 基础入门；

第二篇回顾：Apache Beam实战指南 | 手把手教你玩转KafkaIO与Flink。

关于Apache Beam实战指南系列文章

随着大数据 2.0 时代悄然到来，大数据从简单的批处理扩展到了实时处理、流处理、交互式查询和机器学习应用。近年来涌现出诸多大数据应用组件，如 HBase、Hive、Kafka、Spark、Flink 等。开发者经常要用到不同的技术、框架、API、开发语言和 SDK 来应对复杂应用的开发，这大大增加了选择合适工具和框架的难度，开发者想要将所有的大数据组件熟练运用几乎是一项不可能完成的任务。

面对这种情况，Google 在 2016 年 2 月宣布将大数据流水线产品（Google DataFlow）贡献给 Apache 基金会孵化，2017 年 1 月 Apache 对外宣布开源 Apache Beam，2017 年 5 月迎来了它的第一个稳定版本 2.0.0。在国内，大部分开发者对于 Beam 还缺乏了解，社区中文资料也比较少。InfoQ 期望通过 Apache Beam 实战指南系列文章 推动 Apache Beam 在国内的普及。

一．概述

随着2018年10月2日欧洲Beam首届峰会结束后，Beam的使用者越来越多，关注度越来越高。不光外国公司Google、Spotify、亚马逊、Data Artisans等用上了Beam，TensorFlow机器学习框架也跟Beam结合使用做机器学习的预处理工作，背靠谷歌巨头，Beam不光在大数据一统上做强有力的部署，在云计算、大数据、机器学习、人工智能的集成和运用也越来越广泛。

Beam 在发布第一个版本后，不断完善模型和运行平台。SDKs 也添加了许多IO，例如消息中间件又新增了ActiveMQ 和RabbitMQ ，缓存新增Redis ，大数据分析神器Kudu，大数据存储格式Parquet 等等。Runner 新增了实时流处理Samza和JStorm、MapReduce和加速Hadoop 查询Tez，此外新增了Beam部署Docker 的 DockerCommand 接口，以及Metrics 监控的引入和集成。其他SDK和Runner也在不断更新中，Beam每6周发布一个小版本，及时完善了一些一次性未集成完善的功能。

在科技日新月异的浪潮中，不管是人工智能的机器学习、还是AI的人脸识别、以及物联网的工业互联、互联网的深度挖掘等都必须有一定的数据积累，恰恰这些早期的数据很多公司都存到不同的数据库中，很多公司在早期没有其他大数据存储情况下，基本都存在Hadoop的HDFS中。对于HDFS这个被公认的大数据存储基石，Beam是怎样简单的操作的呢？底层源码是怎样跟Beam结合使用的？我们今天就重点看一下。

二．Apache Beam 中HDFSIO 源码剖析

由于Beam 在发布稳定版本2.0之前的源码，Beam 操作 HDFSIO都比较不稳定，并且API都比较Low。在2.0版本之后HDFSIO的变化很大，2.0版本之前命名为HDFSFileSink读写等操作，2.0之后都是命名为HadoopFileSystem来操作Hadoop 的HDFS 。本文按照Beam 2.4 版本源码进行剖析，2.4之后的版本基本没有很大变化，直到最新的2.9版本才有一个小优化，2.4版本的HDFSIO还是比较稳定的。

2.1 Hdfs的配置类Configuration源码部分

HadoopFileSystem(Configuration configuration) throws IOException { this.fileSystem = org.apache.hadoop.fs.FileSystem.newInstance(configuration);

}

在源码中HadoopFileSystem 把Hadoop Hdfs 的配置类当参数，在构造函数外面配置好传参到内部。 Configuration类其实有三个配置，一个是HdfsConfiguration类，另外是map-reduce Job 任务和YarnConfiguration 资源调度器用到的配置，今天我们主要看HdfsConfiguration 类，因为 Map-Reduce 去年集成到Beam之后基本很少人使用。

在HdfsConfiguration 类中支持很多配置，最主要的“fs.default.name’”是配置我们Hadoop集群。

2.2 HDFS的读写都是基于ByteBuffer的

@Override

public int read(ByteBuffer dst) throws IOException {

if (closed) {

throw new IOException("Channel is closed");

}

// O length read must be supported

int read = 0;

// We avoid using the ByteBuffer based read for Hadoop because some FSDataInputStream

// implementations are not ByteBufferReadable,

// See https://issues.apache.org/jira/browse/HADOOP-14603

if (dst.hasArray()) {
// does the same as inputStream.read(dst):
// stores up to dst.remaining() bytes into dst.array() starting at dst.position().
// But dst can have an offset with its backing array hence the + dst.arrayOffset()

read = inputStream.read(dst.array(), dst.position() + dst.arrayOffset(), dst.remaining());
} else {
// TODO: Add support for off heap ByteBuffers in case the underlying FSDataInputStream
// does not support reading from a ByteBuffer.
read = inputStream.read(dst);
}
if (read > 0) {
dst.position(dst.position() + read);
}

return read;

}

现在支持的外部数据量访问：

1.HarFsInputStream

2.S3InputStream

3.DFSInputStream

4.SwiftNativeInputStream

5.NativeS3FsInputStream

6.LocalFSFileInputStream

7.NativeAzureFsInputStream

8.S3AInputStream

不支持的有：

FTPInputStream

三．Apache Beam 中MySQLIO 源码剖析

MySQL是一款经典的关系型数据库，随着互联网和移动互联网的发展而大放异彩。基本上大大小小公司都在用。本章通过Beam MySQLIO的源码给大家解读一下如何通过Beam来操作MySQL。

MySQL 的集成源码非常简单，就是一个JdbcIO.java 类，为了统一，这里还是看一下Beam 2.4的源码。

3.1 Beam 从MySQL读取全部表数据。

示例：

pipeline.apply(JdbcIO.<KV<Integer, String>>read()

//数据库JDBCIO的配置

.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(

//mysql的驱动

"com.mysql.jdbc.Driver", "jdbc:mysql://hostname:3306/mydb")

//MySQL的用户名

.withUsername("username")

//MySQL的密码

.withPassword("password"))

//查询SQL语句

.withQuery("select id,name from Person")

//返回数据的编码

.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))

//映射返回数据

.withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {

public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {

return KV.of(resultSet.getInt(1), resultSet.getString(2));

}

})

);

MySQL我之前是手工从底层封装过操作类，写法可能跟之前我们之前多少有些区别。实际用的时候大家可以把数据源换成数据库连接池c3p0或阿里巴巴的Druid，其实在Beam JdbcIO.java 源码中，DataSourceConfiguration.create ()方法中有三个重载方法，其中DataSourceConfiguration create(DataSource dataSource) 方法就可以用数据库连接池操作。

大多数操作数据库现在基本是用实体对象通过映射操作的。在源码示例中JdbcIO.RowMapper<KV<Integer, String>>() 的KV<Integer, String> 类型在实际应用中可以换成大家最熟悉的实体对象，因为Beam底层源码中RowMapper 类型是自定义泛型类型T，可以直接定义成实体类型。

3.2 Beam 怎么读取带条件的数据呢？

//添加带条件的SQL读取
.withQuery("select id,name from Person where name = ?")
//数据编码
.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))
//实例化JdbcIO参数类
.withStatementPreparator(new JdbcIO.StatementPreparator() {
//传入执行Statement
public void setParameters(PreparedStatement preparedStatement)
throws Exception {

//传递条件参数
preparedStatement.setString(1, "Darwin");
}
})
.withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {

3.3 Beam 操作Mysql 的批处理。

Batch 的处理条数默认是1000条，这个可以利用withBatchSize()方法进行配置。

/**

* Provide a maximum size in number of SQL statenebt for the batch. Default is 1000.

*
* @param batchSize maximum batch size in number of statements
*/
public Write<T> withBatchSize(long batchSize) {
checkArgument(batchSize > 0, "batchSize must be > 0, but was %d", batchSize);
return toBuilder().setBatchSize(batchSize).build();
}

具体的执行跟其他读取写入返回一样，不过在这里传递的也是一个自定义泛型类型。可以传递我们的SQL List 类型的数据。

private void processRecord(T record, PreparedStatement preparedStatement) {

try {

preparedStatement.clearParameters();
spec.getPreparedStatementSetter().setParameters(record, preparedStatement);
preparedStatement.addBatch();
} catch (Exception e) {

throw new RuntimeException(e);
}
}

MySQL写入在以下示例中已经运用，不再做具体剖析。

四．Apache Beam HDFSIO 实战

很多时候我们要把HDFS批量计算或其他处理到前台展示。我们就写一个从HDFS读取数据，进行分析统计，最后输出到MySQL数据库的实战示例。

4.1 设计思路图

4.2 软件环境和版本说明

系统版本 centos 7
hadoop-2.7.2.tar.gz
mysql-5.7.20-linux-glibc2.12-x86_64.tar.gz
Jdk 1.8
Maven 3.3.3
Spring Tool Suite
spark-2.2.0-bin-hadoop2.7.tgz

4.3 Mysql 数据库设计

CREATE TABLE `test` (

`id` int(11) NOT NULL AUTO_INCREMENT,

`keystr` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,

`count` int(11) DEFAULT NULL,

PRIMARY KEY (`id`)

) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

4.4 实战过程

新建一个Maven项目

引入pom.xml 必要的jar 包。

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-runners-core-java</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>log4j</groupId>

<artifactId>log4j</artifactId>

<version>1.2.17</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-runners-direct-java</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-io-jdbc</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-core</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-io-hadoop-file-system</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-io-hdfs</artifactId>

<version>0.6.0</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-io-hadoop-common</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>jdk.tools</groupId>

<artifactId>jdk.tools</artifactId>

<version>1.8</version>

<scope>system</scope>

<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>

</dependency>

<dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-common</artifactId>

<version>2.7.2</version>

<scope>provided</scope>

</dependency>

<!-- 不加上会报 版本比这是由于Jackson这个工具库的版本不一致导致的

Caused by: com.fasterxml.jackson.databind.JsonMappingException: Incompatible Jackson version:-->

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<version>2.2.0</version>

<exclusions>

<exclusion>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>*</artifactId>

</exclusion>

</exclusions>

</dependency>

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming_2.11</artifactId>

<version>2.2.0</version>

<exclusions>

<exclusion>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>*</artifactId>

</exclusion>

</exclusions>

</dependency>

<dependency>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-core</artifactId>

<version>2.6.0</version>

</dependency>

<dependency>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-databind</artifactId>

<version>2.6.7</version>

</dependency>

<dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-hdfs</artifactId>

<version>2.7.2</version>

</dependency>

<dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-mapreduce-client-core</artifactId>

<version>2.7.2</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-sdks-java-io-jdbc</artifactId>

<version>2.4.0</version>

</dependency>

<!-- 不加上会出现找不到驱动问题 -->

<dependency>

<groupId>mysql</groupId>

<artifactId>mysql-connector-java</artifactId>

<version>5.1.17</version>

</dependency>

<dependency>

<groupId>org.apache.beam</groupId>

<artifactId>beam-runners-spark</artifactId>

<version>2.4.0</version>

</dependency>

<dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

<version>2.7.2</version>

</dependency>

3）新建HTM.java类.

编写以下代码

public static void main(String[] args) {

//配置Hdfs配置

Configuration conf = new Configuration();

//配置Hdfs 的地址

conf.set("fs.default.name", "hdfs://192.168.220.140:9000");

//设置管道

HadoopFileSystemOptions options = PipelineOptionsFactory.fromArgs(a

 

rgs).withValidation().as(HadoopFileSystemOptions.class);

 

options.setHdfsConfiguration(ImmutableList.of(conf));

//这里可以指定 任意平台，我这里本地测试用的本地Runner

options.setRunner(DirectRunner.class);

// options.setRunner(SparkRunner.class);

// options.setRunner(FlinkRunner.class);

Pipeline pipeline = Pipeline.create(options);

// 读取Hdfs的数据跟读取TextIO一样的。

 

PCollection<String> resq = pipeline.apply(TextIO.read().from("hdf

 

s://192.168.220.140:9000/user/lenovo/testfile/test.txt")).apply

 

("ExtractWords",ParDo.of(new DoFn<String, String>()

{

private static final long serialVersionUID = 1L;

 

@ProcessElement

public void processElement(ProcessContext c) {

// 根据空格进行读取数据，里面可以用Luma 表达式写

for (String word : c.element().split(" ")) {

if (!word.isEmpty()) {

c.output(word);

System.out.println(word + "\n");

}

}

}

}));

 

PCollection<String> windowedEvents = resq.apply(Window.<String>

 

into(FixedWindows.of(Duration.standardSeconds(5))));

//把数据出现频率做个统计

PCollection<KV<String, Long>> wordcount = windowedEvents.apply(Count.<String>perElement());

//写入mysql,调用 Beam 的JdbcIO 进行写入

wordcount.apply(JdbcIO.<KV<String, Long>>write()

 

.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration

//mysql 的驱动和连接

.create("com.mysql.jdbc.Driver", "jdbc:mysql://192.168.220.

 

140:3306/db1?useUnicode=true&characterEncoding=UTF-8&allowM ultiQueries=true")

//mysql 的用户名和密码

.withUsername("root").withPassword("123456"))

//SQL语句

.withStatement("insert into test (keystr,count) values(?,?)")

.withPreparedStatementSetter(new JdbcIO.PreparedStatementSe

 

tter<KV<String, Long>>() {

 

private static final long serialVersionUID = 1L;

public void setParameters(KV<String, Long> element, PreparedSta

 

tement query) throws SQLException {

//传递的参数

query.setString(1, element.getKey());

query.setLong(2, element.getValue());

}

}));

pipeline.run().waitUntilFinish();

System.exit(0);

}

5）本地程序运行结果

MySQL数据库执行结果

6）FlinkRunner 执行方式

FlinkRunner 请参照 Apache Beam实战指南 | 手把手教你玩转KafkaIO与Flink

7）SparkRunner 执行方式

首先导出为jar并上传到Llinux spark master 服务器上，如果上了Docker，可以直接用Jenkins构建。

进入 spark bin 目录执行以下命令。

./spark-submit --master spark://master:7077 --class com.BeamHdfsToMySQL.HdfsToMySQL.HTM /home/beamsparkHdfstoMysql.jar

命令解读：

–master spark 通讯地址

–class Beam 的class执行类路径

–/home/beamsparkHdfstoMysql.jar 执行的jar 路径
–runner=SparkRunner 这个是执行执行选择管道用的参数，如果程序未指定则要加上。

注意，如果在程序显示指定 SparkRunner 就不要执行参数再指定，否则会报错。

执行效果如下：

七．总结

不管是人工智能还是物联网，真正大数据的到来才是考验流批计算处理真正的时候的来临。Beam不光在数据源IO上越来越完善，集成越来越多。在计算平台上更新速度也非常快，很多其他社区的开源负责人也在积极封装SDK集成给Beam（例如ClickHouseIO，Nemo Runner），相信以现在的更新速度，Beam用不了多久就可以打造自己的生态。而距离程序员只写一套Beam代码就能在各种计算处理平台上使用的日子也越来越近了。

作者介绍

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

延伸阅读：

《Apache Beam 实战指南一 | 基础入门》
《Apache Beam实战指南二 | 手把手教你玩转KafkaIO与Flink》

创作场景

Apache Beam 实战指南 | 玩转大数据存储 HDFSIO