大數據框架—Flink與Beam

阿新 • • 發佈：2018-04-10

Flink Beam 大數據 WordCount

Flink概述

Flink是Apache的一個頂級項目，Apache Flink 是一個開源的分布式流處理和批處理系統。Flink 的核心是在數據流上提供數據分發、通信、具備容錯的分布式計算。同時，Flink 在流處理引擎上構建了批處理引擎，原生支持了叠代計算、內存管理和程序優化。

現有的開源計算方案，會把流處理和批處理作為兩種不同的應用類型，因為它們所提供的SLA（Service-Level-Aggreement）是完全不相同的：流處理一般需要支持低延遲、Exactly-once保證，而批處理需要支持高吞吐、高效處理。

Flink從另一個視角看待流處理和批處理，將二者統一起來：Flink是完全支持流處理，也就是說作為流處理看待時輸入數據流是無界的；批處理被作為一種特殊的流處理，只是它的輸入數據流被定義為有界的。

Flink流處理特性：

支持高吞吐、低延遲、高性能的流處理
支持帶有事件時間的窗口（Window）操作
支持有狀態計算的Exactly-once語義
支持高度靈活的窗口（Window）操作，支持基於time、count、session，以及data-driven的窗口操作
支持具有Backpressure功能的持續流模型
支持基於輕量級分布式快照（Snapshot）實現的容錯
一個運行時同時支持Batch on Streaming處理和Streaming處理
Flink在JVM內部實現了自己的內存管理
支持叠代計算
支持程序自動優化：避免特定情況下Shuffle、排序等昂貴操作，中間結果有必要進行緩存

Flink架構圖：
技術分享圖片

Flink以層級式系統形式組件其軟件棧，不同層的棧建立在其下層基礎上，並且各層接受程序不同層的抽象形式。

在最基本的層面上，一個Flink應用程序是由以下幾部分組成：

Data source: 數據源，將數據輸入到Flink中
Transformations: 處理數據
Data sink: 將處理後的數據傳輸到某個地方

如下圖：
技術分享圖片

目前Flink支持如下框架：

Apache Kafka (sink/source)
Elasticsearch 1.x / 2.x / 5.x (sink)
HDFS (sink)
RabbitMQ (sink/source)
Amazon Kinesis Streams (sink/source)

Twitter (source)
Apache NiFi (sink/source)
Apache Cassandra (sink)
Redis, Flume, and ActiveMQ (via Apache Bahir) (sink)

Flink官網地址如下：

http://flink.apache.org/

部分內容參考自如下文章：

https://blog.csdn.net/jdoouddm7i/article/details/62039337

使用Flink完成wordcount統計

Flink下載地址：

http://flink.apache.org/downloads.html

Flink快速開始文檔地址：

https://ci.apache.org/projects/flink/flink-docs-release-1.4/quickstart/setup_quickstart.html

註：安裝Flink之前系統中需要安裝有jdk1.7以上版本的環境

我這裏下載的是2.6版本的Flink：

[root@study-01 ~]# cd /usr/local/src/
[root@study-01 /usr/local/src]# wget http://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.4.2/flink-1.4.2-bin-hadoop26-scala_2.11.tgz
[root@study-01 /usr/local/src]# tar -zxvf flink-1.4.2-bin-hadoop26-scala_2.11.tgz -C /usr/local
[root@study-01 /usr/local/src]# cd ../flink-1.4.2/
[root@study-01 /usr/local/flink-1.4.2]# ls
bin  conf  examples  lib  LICENSE  log  NOTICE  opt  README.txt  resources  tools
[root@study-01 /usr/local/flink-1.4.2]#

啟動Flink：

[root@study-01 /usr/local/flink-1.4.2]# ./bin/start-local.sh
[root@study-01 /usr/local/flink-1.4.2]# jps
6576 Jps
6131 JobManager
6499 TaskManager
[root@study-01 /usr/local/flink-1.4.2]#

啟動成功之後就可以訪問主機ip的8081端口，進入到Flink的web頁面：
技術分享圖片

我們現在就可以開始實現wordcount案例了，我這裏有一個文件，內容如下：

[root@study-01 /usr/local/flink-1.4.2]# cat /data/hello.txt 
hadoop welcome
hadoop hdfs mapreduce
hadoop hdfs
hello hadoop
spark vs mapreduce
[root@study-01 /usr/local/flink-1.4.2]#

執行如下命令，實現wordcount案例，如果學習過Hadoop會發現這個命令和Hadoop上使用MapReduce實現wordcount案例是類似的：

[root@study-01 /usr/local/flink-1.4.2]# ./bin/flink run ./examples/batch/WordCount.jar  --input file:///data/hello.txt --output file:///data/tmp/flink_wordcount_out

執行完成後，可以到web頁面上，查看任務的執行信息：
技術分享圖片

查看輸出結果：

[root@study-01 /usr/local/flink-1.4.2]# cat /data/tmp/flink_wordcount_out 
hadoop 4
hdfs 2
hello 1
mapreduce 2
spark 1
vs 1
welcome 1
[root@study-01 /usr/local/flink-1.4.2]#

Beam概述

Google的新老三駕馬車：

老的三駕馬車：GFS、MapReduce、BigTable
新的三駕馬車：Dremel、Pregel、Caffeine

我們都知道，Hadoop生態圈內的幾個框架都源於Google老的三駕馬車，而一些新的框架實現也是部分源於Google新的三駕馬車的概念。所以現在市面上的大數據相關框架很多，框架多就會導致編程規範多、處理模式不一致，而我們希望有一個工具能夠統一這些編程模型，因此，Beam就誕生了。

Apache Beam是 Apache 軟件基金會於2017年1 月 10 日對外宣布的開源平臺。Beam 為創建復雜數據平行處理管道，提供了一個可移動（兼容性好）的 API 層。這層 API 的核心概念基於 Beam 模型（以前被稱為 Dataflow 模型），並在每個 Beam 引擎上不同程度得執行。

背景：

2016 年 2 月份，谷歌及其合作夥伴向 Apache 捐贈了一大批代碼，創立了孵化中的 Beam 項目（最初叫 Apache Dataflow）。這些代碼中的大部分來自於谷歌 Cloud Dataflow SDK——開發者用來寫流處理和批處理管道（pipelines）的庫，可在任何支持的執行引擎上運行。當時，支持的主要引擎是谷歌 Cloud Dataflow，附帶對 Apache Spark 和開發中的 Apache Flink 支持。如今，它正式開放之時，已經有五個官方支持的引擎。除去已經提到的三個，還包括 Beam 模型和 Apache Apex。

Beam特點：

統一了數據批處理（batch）和流處理（stream）編程範式，
能在任何執行引擎上運行。
它不僅為模型設計、更為執行一系列數據導向的工作流提供了統一的模型。這些工作流包括數據處理、吸收和整合。

Beam的官方網站：

https://beam.apache.org/

將WordCount的Beam程序以多種不同Runner運行

Beam Java的快速開始文檔：

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

安裝Beam的前置也是需要系統具備jdk1.7以上版本的環境，以及Maven環境。

使用如下命令下載Beam以及wordcount案例代碼：

mvn archetype:generate       -DarchetypeGroupId=org.apache.beam       -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples       -DarchetypeVersion=2.4.0       -DgroupId=org.example       -DartifactId=word-count-beam       -Dversion="0.1"       -Dpackage=org.apache.beam.examples       -DinteractiveMode=false

進入下載後的目錄進行查看：

[root@study-01 /usr/local/src]# cd word-count-beam/
[root@study-01 /usr/local/src/word-count-beam]# tree
.
├── pom.xml
└── src
    ├── main
    │   └── java
    │       └── org
    │           └── apache
    │               └── beam
    │                   └── examples
    │                       ├── common
    │                       │   ├── ExampleBigQueryTableOptions.java
    │                       │   ├── ExampleOptions.java
    │                       │   ├── ExamplePubsubTopicAndSubscriptionOptions.java
    │                       │   ├── ExamplePubsubTopicOptions.java
    │                       │   ├── ExampleUtils.java
    │                       │   └── WriteOneFilePerWindow.java
    │                       ├── complete
    │                       │   └── game
    │                       │       ├── GameStats.java
    │                       │       ├── HourlyTeamScore.java
    │                       │       ├── injector
    │                       │       │   ├── Injector.java
    │                       │       │   ├── InjectorUtils.java
    │                       │       │   └── RetryHttpInitializerWrapper.java
    │                       │       ├── LeaderBoard.java
    │                       │       ├── StatefulTeamScore.java
    │                       │       ├── UserScore.java
    │                       │       └── utils
    │                       │           ├── GameConstants.java
    │                       │           ├── WriteToBigQuery.java
    │                       │           ├── WriteToText.java
    │                       │           └── WriteWindowedToBigQuery.java
    │                       ├── DebuggingWordCount.java
    │                       ├── MinimalWordCount.java
    │                       ├── WindowedWordCount.java
    │                       └── WordCount.java
    └── test
        └── java
            └── org
                └── apache
                    └── beam
                        └── examples
                            ├── complete
                            │   └── game
                            │       ├── GameStatsTest.java
                            │       ├── HourlyTeamScoreTest.java
                            │       ├── LeaderBoardTest.java
                            │       ├── StatefulTeamScoreTest.java
                            │       └── UserScoreTest.java
                            ├── DebuggingWordCountTest.java
                            ├── MinimalWordCountTest.java
                            └── WordCountTest.java

20 directories, 31 files
[root@study-01 /usr/local/src/word-count-beam]#

默認情況下，beam的runner是Direct，下面就用Direct來運行wordcount案例，命令如下：

[root@study-01 /usr/local/src/word-count-beam]# ls
pom.xml  src  target
[root@study-01 /usr/local/src/word-count-beam]#
[root@study-01 /usr/local/src/word-count-beam]# mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--inputFile=/data/hello.txt --output=counts" -Pdirect-runner

運行的結果會存放在當前的目錄下：

[root@study-01 /usr/local/src/word-count-beam]# ls
counts-00000-of-00003  counts-00001-of-00003  counts-00002-of-00003  pom.xml  src  target
[root@study-01 /usr/local/src/word-count-beam]# more counts*  # 查看結果文件
::::::::::::::
counts-00000-of-00003
::::::::::::::
welcome: 1
spark: 1
::::::::::::::
counts-00001-of-00003
::::::::::::::
hdfs: 2
hadoop: 4
mapreduce: 2
::::::::::::::
counts-00002-of-00003
::::::::::::::
hello: 1
vs: 1
[root@study-01 /usr/local/src/word-count-beam]#

如果需要指定其他的runner則可以使用--runner參數進行指定，例如我要指定runner為Flink，則修改命令如下即可：

[root@study-01 /usr/local/src/word-count-beam]# mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--runner=FlinkRunner --inputFile=/data/hello.txt --output=counts" -Pflink-runner

刪除之前生成的文件及目錄，我們來使用Spark的方式進行運行。使用Spark的話，也只是修改--runner以及-Pspark參數即可：

[root@study-01 /usr/local/src/word-count-beam]# mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--runner=SparkRunner --inputFile=/data/hello.txt --output=counts" -Pspark-runner

運行成功後，也是會生成如下文件及目錄：

[root@study-01 /usr/local/src/word-count-beam]# ls
counts-00000-of-00003  counts-00001-of-00003  counts-00002-of-00003  pom.xml  src  target
[root@study-01 /usr/local/src/word-count-beam]#

查看處理結果：

[root@study-01 /usr/local/src/word-count-beam]# more counts*
::::::::::::::
counts-00000-of-00003
::::::::::::::
spark: 1
::::::::::::::
counts-00001-of-00003
::::::::::::::
welcome: 1
hello: 1
mapreduce: 2
::::::::::::::
counts-00002-of-00003
::::::::::::::
vs: 1
hdfs: 2
hadoop: 4
[root@study-01 /usr/local/src/word-count-beam]#

以上這兩個示例只是想說明一點，同一份代碼，可以運行在不同的計算引擎上。不需要為不同的引擎開發不同的代碼，這就是Beam框架的最主要的設計目的之一。

大數據框架—Flink與Beam

Flink Beam 大數據 WordCount Flink概述 Flink是Apache的一個頂級項目，Apache Flink 是一個開源的分布式流處理和批處理系統。Flink 的核心是在數據流上提供數據分發、通信、具備容錯的分布式計算。同時，Flink 在流處理引擎上構建了批處理引擎，原

大數據框架—Flink與Beam

使用Flink完成wordcount統計

Beam概述

將WordCount的Beam程序以多種不同Runner運行

大數據框架—Flink與Beam

hbase 表的設計與其它大數據框架的集成

運維監控大數據的提取與分析

大數據-菜鳥與順豐互撕的背後

PHP解決網站大數據大流量與高並發

小白學習大數據測試之揭秘大數據的背景與發展

什麽是雲計算？大數據、AI與雲計算有什麽區別？

hadoop大數據處理平臺與案例

給Java開發者的十個大數據框架和工具

大數據技術原理與應用——大數據處理架構Hadoop

大數據技術原理與應用——分布式文件系統HDFS

大數據分析技術與應用

新手必備：大數據框架Hadoop主要模塊解析

【Flink】流計算框架Flink與Storm的效能對比

流計算框架 Flink 與 Storm 的效能對比

拉開大變革序幕（下）：分布式計算框架與大數據

一文讀懂大數據計算框架與平臺

分享《深度學習與計算機視覺算法原理框架應用》《大數據架構詳解從數據獲取到深度學習》PDF數據集

【大數據開發】你知道大數據語言的工具與框架嗎？

Spark與Flink大數據處理引擎對比分析！

大數據框架—Flink與Beam

使用Flink完成wordcount統計

Beam概述

將WordCount的Beam程序以多種不同Runner運行

相關推薦