Apache Beam 剖析

1.概述

在大数据的浪潮之下，技术的更新迭代十分频繁。受技术开源的影响，大数据开发者提供了十分丰富的工具。但也因为如此，增加了开发者选择合适工具的难度。在大数据处理一些问题的时候，往往使用的技术是多样化的。这完全取决于业务需求，比如进行批处理的MapReduce，实时流处理的Flink，以及SQL交互的Spark SQL等等。而把这些开源框架，工具，类库，平台整合到一起，所需要的工作量以及复杂度，可想而知。这也是大数据开发者比较头疼的问题。而今天要分享的就是整合这些资源的一个解决方案，它就是 Apache Beam。

2.内容

Apache Beam 最初叫 Apache Dataflow，由谷歌和其合作伙伴向Apache捐赠了大量的核心代码，并创立孵化了该项目。该项目的大部分大码来自于 Cloud Dataflow SDK，其特点有以下几点：

  • 统一数据批处理(Batch)和流处理(Stream)编程的范式
  • 能运行在任何可执行的引擎之上

那 Apache Beam到底能解决哪些问题，它的应用场景是什么，下面我们可以通过一张图来说明，如下图所示：

通过改图，我们可以很清晰的看到整个技术的发展流向;一部分是谷歌派系，另一部分则是Apache派系。在开发大数据应用时，我们有时候使用谷歌的框架，API，类库，平台等，而有时候我们则使用Apache的，比如：HBase，Flink，Spark等。而我们要整合这些资源则是一个比较头疼的问题，Apache Beam 的问世，整合这些资源提供了很方便的解决方案。

2.1 Vision

下面，我们通过一张流程图来看Beam的运行流程，如下图所示：

通过上图，我们可以清楚的知道，执行一个流程分以下步骤：

  1. End Users：选择一种你熟悉的编程语言提交应用
  2. SDK Writers：该编程语言必须是 Beam 模型支持的
  3. Library Writers：转换成Beam模型的格式
  4. Runner Writers：在分布式环境下处理并支持Beam的数据处理管道
  5. IO Providers：在Beam的数据处理管道上运行所有的应用
  6. DSL Writers：创建一个高阶的数据处理管道

2.2 SDK

Beam SDK 提供了一个统一的编程模型，来处理任意规模的数据集，其中包括有限的数据集，无限的流数据。Apache Beam SDK 使用相同的类来表达有限和无限的数据，同样使用相同的转换方法对数据进行操作。Beam 提供了多种 SDK，你可以选择一种你熟悉的来建立数据处理管道，如上述的 2.1 中的图，我们可以知道，目前 Beam 支持 Java，Python 以及其他待开发的语言。

2.3 Pipeline Runners

在 Beam 管道上运行引擎会根据你选择的分布式处理引擎，其中兼容的 API 转换你的 Beam 程序应用，让你的 Beam 应用程序可以有效的运行在指定的分布式处理引擎上。因而，当运行 Beam 程序的时候，你可以按照自己的需求选择一种分布式处理引擎。当前 Beam 支持的管道运行引擎有以下几种：

  • Apache Apex
  • Apache Flink
  • Apache Spark
  • Google Cloud Dataflow

3.示例

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

3.1 开发环境

  • 下载安装 JDK 7 或更新的版本，检测 JAVA_HOME环境变量
  • 下载 Maven 打包环境。

关于上述的安装步骤，并不是本篇博客的重点，这里笔者就不多赘述了，不明白的可以到官网翻阅文档进行安装。

3.2 下载示例代码

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

然后，将其中的示例代码进行打包，命令如下所示：



    

  1. $ mvn archetype:generate \ 
    2. 

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

  3.       -DarchetypeGroupId=org.apache.beam \ 
    4. 

  4.       -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \ 
    5. 

  5.       -DarchetypeVersion=LATEST \ 
    6. 

  6.       -DgroupId=org.example \ 
    7. 

  7.       -DartifactId=word-count-beam \ 
    8. 

  8.       -Dversion="0.1" \ 
    9. 

  9.       -Dpackage=org.apache.beam.examples \ 
    10. 

  10.       -DinteractiveMode=false 
    11. 



此时，命令会创建一个文件夹 word-count-beam，里面包含一个 pom.xml 和相关的代码文件。命令如下所示：




    

  1. $ cd word-count-beam/ 
    2. 

  2.  
    3. 

  3. $ ls 
    4. 

  4. pom.xml    src 
    5. 

  5.  
    6. 

  6. $ ls src/main/java/org/apache/beam/examples/ 
    7. 

  7. DebuggingWordCount.java    WindowedWordCount.java    common 
    8. 

  8. MinimalWordCount.java    WordCount.java 
    9. 



3.3 运行 WordCount 示例代码


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


下面，你可以按需选择你想执行程序的引擎：


    

  1. 对引擎进行相关配置
    2. 

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

  3. 运行示例程序
    4. 



3.3.1 Direct




    

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

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



3.3.2 Apex




    

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

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



3.3.3 Flink-Local




    

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

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



3.3.4 Flink-Cluster




    

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

  2. -Dexec.args="--runner=FlinkRunner --flinkMaster=<flink master> --filesToStage=target/word-count-beam-bundled-0.1.jar \  
    3. 

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



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


3.3.5 Spark




    

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

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



3.3.6 Dataflow




    

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

  2.      -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://<your-gcs-bucket>/tmp \ 
    3. 

  3.                   --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://<your-gcs-bucket>/counts" \ 
    4. 

  4.      -Pdataflow-runner 
    5. 



3.4 运行结果


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


3.4.1 Direct




    

  1. $ ls counts* 
    2. 

  2.  
    3. 

  3. $ more counts* 
    4. 

  4. api: 9 
    5. 

  5. bundled: 1 
    6. 

  6. old: 4 
    7. 

  7. Apache: 2 
    8. 

  8. The: 1 
    9. 

  9. limitations: 1 
    10. 

  10. Foundation: 1 
    11. 

  11. ... 
    12. 



3.4.2 Apex




    

  1. $ cat counts* 
    2. 

  2. BEAM: 1 
    3. 

  3. have: 1 
    4. 

  4. simple: 1 
    5. 

  5. skip: 4 
    6. 

  6. PAssert: 1 
    7. 

  7. ... 
    8. 



3.4.3 Flink-Local




    

  1. $ ls counts* 
    2. 

  2.  
    3. 

  3. $ more counts* 
    4. 

  4. The: 1 
    5. 

  5. api: 9 
    6. 

  6. old: 4 
    7. 

  7. Apache: 2 
    8. 

  8. limitations: 1 
    9. 

  9. bundled: 1 
    10. 

  10. Foundation: 1 
    11. 

  11. ... 
    12. 



3.4.4 Flink-Cluster




    

  1. $ ls /tmp/counts* 
    2. 

  2.  
    3. 

  3. $ more /tmp/counts* 
    4. 

  4. The: 1 
    5. 

  5. api: 9 
    6. 

  6. old: 4 
    7. 

  7. Apache: 2 
    8. 

  8. limitations: 1 
    9. 

  9. bundled: 1 
    10. 

  10. Foundation: 1 
    11. 

  11. ... 
    12. 



3.4.5 Spark




    

  1. $ ls counts* 
    2. 

  2.  
    3. 

  3. $ more counts* 
    4. 

  4. beam: 27 
    5. 

  5. SF: 1 
    6. 

  6. fat: 1 
    7. 

  7. job: 1 
    8. 

  8. limitations: 1 
    9. 

  9. require: 1 
    10. 

  10. of: 11 
    11. 

  11. profile: 10 
    12. 

  12. ... 
    13. 



3.4.6 Dataflow




    

  1. $ gsutil ls gs://<your-gcs-bucket>/counts* 
    2. 

  2.  
    3. 

  3. $ gsutil cat gs://<your-gcs-bucket>/counts* 
    4. 

  4. feature: 15 
    5. 

  5. smother'st: 1 
    6. 

  6. revelry: 1 
    7. 

  7. bashfulness: 1 
    8. 

  8. Bashful: 1 
    9. 

  9. Below: 2 
    10. 

  10. deserves: 32 
    11. 

  11. barrenly: 1 
    12. 

  12. ... 
    13. 



4.总结


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


本文作者：佚名


来源：51CTO
				


				
时间： 2025-01-31 01:26:09

		
		


		


	

	
	

		


		
Apache Beam 剖析的相关文章


								

		

			

                Apache Beam研究报告
			

		

		

									

				概述 本文不是一篇Beam的入门文档,不会介绍Beam的基本概念:而会主要探讨Beam的表达力,Beam的性能,以及Beam目前在业内的使用情况.面向的读者是那些想使用Beam作为自己公司操作大数据的统一API,但是还有所顾虑的人们. 表达力 离线 Beam里面有两个核心原语: ParDo: 来处理通用的基于单条数据的计算: 每条需要处理的数据会被喂给用户提供的指定的一个函数(Beam里面的@ProcessElement), 然后输出0个或者多个输出. 我们平常熟悉的Filter, Append			

		

	

				

		

			

                Apache Beam WordCount编程实战及源码解读
			

		

		

									

				概述:Apache Beam WordCount编程实战及源码解读,并通过intellij IDEA和terminal两种方式调试运行WordCount程序,Apache Beam对大数据的批处理和流处理,提供一套先进的统一的编程模型,并可以运行大数据处理引擎上.完整项目Github源码 负责公司大数据处理相关架构,但是具有多样性,极大的增加了开发成本,急需统一编程处理,Apache Beam,一处编程,处处运行,故将折腾成果分享出来. 1.Apache Beam编程实战–前言,Apache B			

		

	

				

		

			

                Apache Beam: 下一代的大数据处理标准
			

		

		

									

				Apache Beam(原名Google DataFlow)是Google在2016年2月份贡献给Apache基金会的Apache孵化项目,被认为是继MapReduce,GFS和BigQuery等之后,Google在大数据处理领域对开源社区的又一个非常大的贡献.Apache Beam的主要目标是统一批处理和流处理的编程范式,为无限,乱序,web-scale的数据集处理提供简单灵活,功能丰富以及表达能力十分强大的SDK.Apache Beam项目重点在于数据处理的编程范式和接口定义,并不涉及具体执			

		

	

				

		

			

                Apache Beam初探
			

		

		

									

				Apache Beam Apache Beam provides an advanced unified programming model, allowing you to implement batch and streaming data processing jobs that can run on any execution engine. Apache Beam is: UNIFIED - Use a single programming model for both batch a			

		

	

				

		

			

                Apache Beam欲通过uber api获取大数据
			

		

		

									

				文章讲的是Apache Beam欲通过uber api获取大数据,现在,有用的Apache大数据项目似乎每日更新.相比于每次都重新学习的方式,如果可以通过一个统一的API如何呢? 长期开玩笑说Hadoop生态系统是那种如果你不喜欢一个为特定系统的API,等待五分钟,两个新的Apache项目将出现随之而来崭新的API可供学习. 有很多要赶着学习.更糟糕的是,它会导致很多工作迁移到不同的项目仅仅为了保持通用性."我们已经在暴风雨中实现了流媒体解决方案!现在我们已经快速地重做了!我们目前正在重写pac			

		

	

				

		

			

                Apache Beam发布第一个稳定版本
			

		

		

									

				Apache Beam在官方博客上正式发布了Beam 2.0.0.这是Beam有史以来的第一个稳定版本,根据Beam社区的声明,Beam意欲为未来版本发布保持API的稳定性,并让Beam适用于企业的部署. Beam的第一个稳定版本是Beam社区发布的第三个重要里程碑.Beam在2016年2月成为Apache孵化器项目,并在同年12月升级成为Apache基金会的顶级项目.经过从开始至今的15个月全神贯注的努力,从一个有点混乱的代码库开始,从各大组织合并代码,成就了如今的这个数据处理框架,它是一个真			

		

	

				

		

			

                大数据统一的批处理和流处理标准 Apache Beam
			

		

		

									

				Apache Beam 是 Apache 软件基金会越来越多的数据流项目中最新增添的成员,是 Google 在2016年2月份贡献给 Apache 基金会的孵化项目. 这个项目的名称表明了设计:结合了批处理(Batch)模式和数据流(Stream)处理模式.它基于一种统一模式,用于定义和执行数据并行处理管道(pipeline),这些管理随带一套针对特定语言的SDK用于构建管道,以及针对特定运行时环境的Runner用于执行管道. Apache Beam 的主要目标是统一批处理和流处理的编程范式,为			

		

	

				

		

			

                2016美国QCon看法：在Beam上，我为什么说Google有统一流式计算的野心
			

		

		

									

				编者按:流式计算(Stream Processing)在经历了若干年的发展之后,已经有了比较完整的生态,如开源的Storm, Flink, Spark等,未开源的如Google的DataFlow,几乎每个巨头都有自己的流式计算系统.生态虽繁荣但分散,各个平台之间也是互不兼容的,一个平台上写的程序很难移植到另外一个平台,这些领域难题再加上Google大一统流式计算的野心催生了Apache孵化器的新项目Beam.            Google是最早实践大数据的公司,目前大数据繁荣的生态很大一部			

		

	

				

		

			

                Alibaba RocketMQ捐赠给Apache那些鲜为人知的故事
			

		

		

									

				序言 今年的双十一对阿里巴巴中间件消息团队来说,注定是个不平凡的日子.在这一天,稳定性小组重点攻克的低延迟存储解决方案成功地经受住了大考.整个大促期间,99.996%的延迟落在了10ms以内,极个别由于GC引发的停顿在50ms以内,对于读写比例几乎均衡的分布式消息引擎来说,这一结果无不令人兴奋.甚至可以毫不夸张地讲,即便拿到明年的Java one大会上,也必定是场非常吸睛的技术干货分享.接下来,团队同学会把相关的经验提炼总结出来,Ata上已经为大家呈现了部分精彩内容,期待能在接下来全球Qcon大