当前位置： 首页 > news >正文

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

news 来源：原创 2024/4/28 6:53:05

本文是 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之后基本很少人使用。

\"image\"

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

2.2 HDFS的读写都是基于ByteBuffer的

@Overridepublic int read(ByteBuffer dst) throws IOException {if (closed) {throw new IOException(\u0026quot;Channel is closed\u0026quot;);}// O length read must be supportedint 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-14603if (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 \u0026gt; 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.\u0026lt;KV\u0026lt;Integer, String\u0026gt;\u0026gt;read()//数据库JDBCIO的配置.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(//mysql的驱动\u0026quot;com.mysql.jdbc.Driver\u0026quot;, \u0026quot;jdbc:mysql://hostname:3306/mydb\u0026quot;)//MySQL的用户名.withUsername(\u0026quot;username\u0026quot;)//MySQL的密码.withPassword(\u0026quot;password\u0026quot;))//查询SQL语句.withQuery(\u0026quot;select id,name from Person\u0026quot;)//返回数据的编码.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))//映射返回数据.withRowMapper(new JdbcIO.RowMapper\u0026lt;KV\u0026lt;Integer, String\u0026gt;\u0026gt;() {public KV\u0026lt;Integer, String\u0026gt; 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) 方法就可以用数据库连接池操作。

\"image\"

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

\"image\"

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

//添加带条件的SQL读取.withQuery(\u0026quot;select id,name from Person where name = ?\u0026quot;)//数据编码.withCoder(KvCoder.of(BigEndianIntegerCoder.of(), StringUtf8Coder.of()))//实例化JdbcIO参数类.withStatementPreparator(new JdbcIO.StatementPreparator() {//传入执行Statementpublic void setParameters(PreparedStatement preparedStatement)throws Exception {//传递条件参数preparedStatement.setString(1, \u0026quot;Darwin\u0026quot;);}}).withRowMapper(new JdbcIO.RowMapper\u0026lt;KV\u0026lt;Integer, String\u0026gt;\u0026gt;() {

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\u0026lt;T\u0026gt; withBatchSize(long batchSize) {checkArgument(batchSize \u0026gt; 0, \u0026quot;batchSize must be \u0026gt; 0, but was %d\u0026quot;, 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 设计思路图

\"image\"

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 实战过程

  1. 新建一个Maven项目

\"image\"

  1. 引入pom.xml 必要的jar 包。
\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-runners-core-java\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;log4j\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;log4j\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;1.2.17\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-runners-direct-java\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-io-jdbc\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-core\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-io-hadoop-file-system\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-io-hdfs\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;0.6.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-io-hadoop-common\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;jdk.tools\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;jdk.tools\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;1.8\u0026lt;/version\u0026gt;\u0026lt;scope\u0026gt;system\u0026lt;/scope\u0026gt;\u0026lt;systemPath\u0026gt;${JAVA_HOME}/lib/tools.jar\u0026lt;/systemPath\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.hadoop\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;hadoop-common\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.7.2\u0026lt;/version\u0026gt;\u0026lt;scope\u0026gt;provided\u0026lt;/scope\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;!-- 不加上会报 版本比这是由于Jackson这个工具库的版本不一致导致的Caused by: com.fasterxml.jackson.databind.JsonMappingException: Incompatible Jackson version:--\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.spark\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;spark-core_2.11\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.2.0\u0026lt;/version\u0026gt;\u0026lt;exclusions\u0026gt;\u0026lt;exclusion\u0026gt;\u0026lt;groupId\u0026gt;com.fasterxml.jackson.core\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;*\u0026lt;/artifactId\u0026gt;\u0026lt;/exclusion\u0026gt;\u0026lt;/exclusions\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.spark\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;spark-streaming_2.11\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.2.0\u0026lt;/version\u0026gt;\u0026lt;exclusions\u0026gt;\u0026lt;exclusion\u0026gt;\u0026lt;groupId\u0026gt;com.fasterxml.jackson.core\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;*\u0026lt;/artifactId\u0026gt;\u0026lt;/exclusion\u0026gt;\u0026lt;/exclusions\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;com.fasterxml.jackson.core\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;jackson-core\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.6.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;com.fasterxml.jackson.core\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;jackson-databind\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.6.7\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.hadoop\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;hadoop-hdfs\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.7.2\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.hadoop\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;hadoop-mapreduce-client-core\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.7.2\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-sdks-java-io-jdbc\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;!-- 不加上会出现找不到驱动问题 --\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;mysql\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;mysql-connector-java\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;5.1.17\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.beam\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;beam-runners-spark\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.4.0\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;\u0026lt;dependency\u0026gt;\u0026lt;groupId\u0026gt;org.apache.hadoop\u0026lt;/groupId\u0026gt;\u0026lt;artifactId\u0026gt;hadoop-client\u0026lt;/artifactId\u0026gt;\u0026lt;version\u0026gt;2.7.2\u0026lt;/version\u0026gt;\u0026lt;/dependency\u0026gt;

3）新建HTM.java类.

\"image\"

  1. 编写以下代码
public static void main(String[] args) {//配置Hdfs配置Configuration conf = new Configuration();//配置Hdfs 的地址conf.set(\u0026quot;fs.default.name\u0026quot;, \u0026quot;hdfs://192.168.220.140:9000\u0026quot;);//设置管道HadoopFileSystemOptions options = PipelineOptionsFactory.fromArgs(a rgs).withValidation().as(HadoopFileSystemOptions.class); options.setHdfsConfiguration(ImmutableList.of(conf));//这里可以指定 任意平台，我这里本地测试用的本地Runneroptions.setRunner(DirectRunner.class);// options.setRunner(SparkRunner.class);// options.setRunner(FlinkRunner.class);Pipeline pipeline = Pipeline.create(options);// 读取Hdfs的数据跟读取TextIO一样的。 PCollection\u0026lt;String\u0026gt; resq = pipeline.apply(TextIO.read().from(\u0026quot;hdf s://192.168.220.140:9000/user/lenovo/testfile/test.txt\u0026quot;)).apply (\u0026quot;ExtractWords\u0026quot;,ParDo.of(new DoFn\u0026lt;String, String\u0026gt;(){private static final long serialVersionUID = 1L; @ProcessElementpublic void processElement(ProcessContext c) {// 根据空格进行读取数据，里面可以用Luma 表达式写for (String word : c.element().split(\u0026quot; \u0026quot;)) {if (!word.isEmpty()) {c.output(word);System.out.println(word + \u0026quot;\\u0026quot;);}}}})); PCollection\u0026lt;String\u0026gt; windowedEvents = resq.apply(Window.\u0026lt;String\u0026gt; into(FixedWindows.of(Duration.standardSeconds(5))));//把数据出现频率做个统计PCollection\u0026lt;KV\u0026lt;String, Long\u0026gt;\u0026gt; wordcount = windowedEvents.apply(Count.\u0026lt;String\u0026gt;perElement());//写入mysql,调用 Beam 的JdbcIO 进行写入wordcount.apply(JdbcIO.\u0026lt;KV\u0026lt;String, Long\u0026gt;\u0026gt;write() .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration//mysql 的驱动和连接.create(\u0026quot;com.mysql.jdbc.Driver\u0026quot;, \u0026quot;jdbc:mysql://192.168.220. 140:3306/db1?useUnicode=true\u0026amp;characterEncoding=UTF-8\u0026amp;allowM ultiQueries=true\u0026quot;)//mysql 的用户名和密码.withUsername(\u0026quot;root\u0026quot;).withPassword(\u0026quot;123456\u0026quot;))//SQL语句.withStatement(\u0026quot;insert into test (keystr,count) values(?,?)\u0026quot;).withPreparedStatementSetter(new JdbcIO.PreparedStatementSe tter\u0026lt;KV\u0026lt;String, Long\u0026gt;\u0026gt;() { private static final long serialVersionUID = 1L;public void setParameters(KV\u0026lt;String, Long\u0026gt; element, PreparedSta tement query) throws SQLException {//传递的参数query.setString(1, element.getKey());query.setLong(2, element.getValue());}}));pipeline.run().waitUntilFinish();System.exit(0);}

5）本地程序运行结果

\"image\"

MySQL数据库执行结果

\"image\"

6）FlinkRunner 执行方式

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

7）SparkRunner 执行方式

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

\"image\"

进入 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 就不要执行参数再指定，否则会报错。

执行效果如下：

\"image\"
\"image\"

七．总结

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

作者介绍

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

延伸阅读：

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

相关文章：

  • 记一次面试题——call、apply、bind模拟实现的更好方式
  • 逻辑运算符
  • 古郡敦煌迎新年初雪 雪漠风光引游人
  • 台湾大学生在威海研习中华文化 感叹收获太多“惊喜”
  • 如何使用 Druid 和 Kafka 构造 Kappa 架构完成流量分析
  • 利用位运算实现加减乘除
  • IT应该自动化的7件事
  • 陕西彬州一男子持刀杀害两名女性 警方发布协查通告
  • 圆方圆：python的错误处理——try语句
  • 洛谷 P1824 【进击的奶牛】
  • 自学自用 = B站（操作系统_清华大学(向勇、陈渝)）1
  • Docker下部署自己的LNMP工作环境
  • Docker-01-使用镜像
  • C# 分割字符串
  • LDM和STM指令
  • 07.Android之多媒体问题
  • bearychat的java client
  • CAP 一致性协议及应用解析
  • Codepen 每日精选（2018-3-25）
  • Hibernate最全面试题
  • JavaWeb(学习笔记二)
  • OpenStack安装流程（juno版）- 添加网络服务（neutron）- controller节点
  • Quartz初级教程
  • spring boot 整合mybatis 无法输出sql的问题
  • WebSocket使用
  • 不上全站https的网站你们就等着被恶心死吧
  • 从0实现一个tiny react（三）生命周期
  • 看完九篇字体系列的文章，你还觉得我是在说字体？
  • 驱动程序原理
  • 如何在 Tornado 中实现 Middleware
  • 我这样减少了26.5M Java内存！
  • 新手搭建网站的主要流程
  • 鱼骨图 - 如何绘制？
  • 云大使推广中的常见热门问题
  • Linux权限管理(week1_day5)--技术流ken
  • 昨天1024程序员节，我故意写了个死循环~
  • ​软考-高级-系统架构设计师教程（清华第2版）【第12章 信息系统架构设计理论与实践（P420~465）-思维导图】​
  • ​一帧图像的Android之旅 :应用的首个绘制请求
  • #【QT 5 调试软件后，发布相关：软件生成exe文件 + 文件打包】
  • #vue3 实现前端下载excel文件模板功能
  • #Z2294. 打印树的直径
  • #常见电池型号介绍 常见电池尺寸是多少【详解】
  • #每日一题合集#牛客JZ23-JZ33
  • #我与Java虚拟机的故事#连载06：收获颇多的经典之作
  • (2/2) 为了理解 UWP 的启动流程，我从零开始创建了一个 UWP 程序
  • （DenseNet）Densely Connected Convolutional Networks--Gao Huang
  • (html转换)StringEscapeUtils类的转义与反转义方法
  • （pytorch进阶之路）扩散概率模型
  • （附源码）springboot高校宿舍交电费系统 毕业设计031552
  • （接口自动化）Python3操作MySQL数据库
  • （蓝桥杯每日一题）love
  • （篇九）MySQL常用内置函数
  • （十一）图像的罗伯特梯度锐化
  • （未解决）macOS matplotlib 中文是方框
  • (一)Neo4j下载安装以及初次使用