大數據數據流組件選擇
作者：尹正傑
版權聲明：原創作品，謝絕轉載！否則將追究法律責任。
一.大數據數據流的架構和組件介紹
1>.什么是數據流
所謂數據流（流數據），是一組順序、大量、快速、連續到達的數據序列，一般情況下，數據流可被視為一個隨時間延續而無限增長的動態數據集合。應用於網絡監控、傳感器網絡、航空航天、氣象測控和金融服務等領域。
流數據具有四個特點： （1）數據實時到達； （2）數據到達次序獨立，不受應用系統所控制； （3）數據規模宏大且不能預知其最大值； （4）數據一經處理，除非特意保存，否則不能被再次取出處理，或者再次提取數據代價昂貴。
2>.大數據架構：Lambda
Lambda Architecture（簡稱：LA）最早是Twitter工程師Nathan Marz提出來的，它是一種大數據軟件設計設計架構，其目的是指導用戶充分利用批處理和流式計算技術各自的優點實現一個復雜的大數據處理系統。通過結合這兩類計算技術，LA可以在延遲，吞吐量和容錯之間找到平衡點。如下圖所示，LA主要思想是將數據處理流分解成三層：批處理層，流式處理層和服務層。
一.批處理層
它的主要思想是利用分布式處理計算，以批處理為單位處理數據，並產生一個經預算產生的只讀數據視圖。該層將數據流看成只讀的，僅支持追加操作的超大數據集。它可以一次性處理大量數據，引入復雜的計算邏輯（比如機器學習中的模型迭代計算，歷史庫的匹配等），其優點是吞吐率高，缺點是數據處理延遲高，即從數據產生到最終被處理完成，整個過程用時較長，通常是分鍾或小時級別。
二.流式處理層
為了降低批處理層帶來的高延遲，LA又引入了流式處理層，該層采用流式計算技術，大大降低了數據處理延遲（通常是毫秒或秒級別），其優點是數據處理延遲低，缺點是無法進行復雜的邏輯計算，得到的結果往往是近似解。
三.服務層
批處理層和流式處理層可以看結合在一起，這樣既保證數據延遲低，也能完成復雜的邏輯計算（只能保證最終一致性）。為了整合兩層的計算結果，LA進一步引入服務層，它對外提供了統一的訪問接口以方便用戶使用。
四.LA應用案例
一個經典的LA應用案例是推薦系統。在互聯網行業，推薦系統被應用在各個領域，包括電子商務，視頻，新聞等。推薦系統等設計目的是根據用戶的興趣特點和購買行為，向用戶推薦感興趣的信息和商品。推薦系統是建立在海量數據挖掘的基礎上的一種高級商務智能平台，以幫助商家為其顧客購物提供完全個性化的決策支持和信息服務。推薦系統最核心的模塊是推薦算法，推薦算法通常會根據用戶的興趣特點和歷史行為數據的構建推薦模型，以預測用戶可能感興趣的信息和商品，進而推薦給用戶。
如下圖所示，它為一個典型的推薦系統數據流水線架構。在該架構中，數據統一流入Kafka，之后按照不同時間粒度導入批處理和流式處理兩個系統中。批處理層擁有所有歷史數據（通常保存到HDFS/HBase中），通常用以實時推薦模型，它以當前數據（比如最近一小時數據）和歷史數據為輸入，通過特征工程，模型構建（通常是迭代算法，使用MapReduce/Spark實現）及模型評估等計算環境后，最終獲得最優模型並將產生的推薦結果存儲（比如Redis）起來，整個過程延遲較大（分鍾甚至小時級別）；為了解決推薦系統中的冷啟動問題（新用戶推薦問題），往往會引入流處理層：他會試試手機用戶的行為，並基於這些行為數據通過簡單推薦算法（通常使用Storm/Spark Streaming實現）快速產生推薦結果並存儲起來。為了便於其他系統獲取推薦結果，推薦系統往往通過服務層對外提供訪問接口，比如網站后台在渲染某個訪問頁面時，可能從廣告系統，推薦系統以及內容存儲系統中獲取對應的結果，並返回給客戶端。
3>.批處理和流處理的比較
批處理一般用於計算從所含的所有數據得到的結果，並實現對大數據集的深入分析。相反，流處理則需要攝取一個數據序列，增量式更新指標、報告和匯總統計結果，以響應每個到達的數據記錄。這種處理方法更適合實時監控和響應調用。
接下來我們從下面幾個維度來分析一下批處理和流處理的區別：
（1）數據范圍
批處理對數據集中的所有或大部分數據進行查詢或處理。流處理對滾動時間窗口內的數據或僅對最近的數據記錄進行查詢或處理。
（2）數據大小
批處理針對的是大批量數據（如GB或者PB級別）。流處理針對的是單條記錄或包含幾條記錄的微批數據（如KB或者MB）。
（3）性能
批處理所需的時間一般是幾分鍾至幾小時的延遲。流處理所需的時間幾毫秒至幾秒的延遲。
（4）場景
批處理使用的場景分析起來很復雜。流處理只需要簡單的響應調用，聚合和滾動指標。
很多企業結合使用兩種方法，從而構建一種混合模式，並同時維持實時處理層和批處理層。
4>.大數據數據流典型架構
數據采集：
負責從各種數據源實時采集數據，在采集時，可能對數據做簡單的ETL或格式轉換，以便於下游系統使用。例如：Apache Flume
數據接入：
由於數據采集的速度和數據處理的速度不一定同步，通常會引入一個消息中間件來作為緩沖。例如：Apache Kafka
流式計算：
對流數據進行實時的處理和分析。例如：Apache Storm
批處理計算：
對大量數據進行離線處理分析，例如Apache MapReduce
數據存儲：
對處理后的結果數據進行保存，以便下游系統進行查詢或再次處理。例如：Apache HBase
二.數據攝取組件
1>.Apache Flume
一.什么是Flume Flume是一個分布式、可靠、高可用的海量日志聚合系統，支持在系統中定制各類數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各種數據接收方。 官方地址：http://flume.apache.org/。 二.Flume特性 （1）高可靠性 Flume提供了end to end的數據可靠性機制 （2）易於擴展 Agent為分布式架構，可水平擴展 （3）易於恢復 Channel中保存了與數據源有關的事件，用於失敗時的恢復 （4）功能豐富 Flume內置了多種組件，包括不同數據源和不同存儲方式 三.Flume常用組件 （1）Source： 數據源，簡單的說就是agent獲取數據的入口。 （2）Channel： 管道，數據流通和存儲的通道。一個source必須至少和一個channel關聯。 （3）Sink： 用來接收channel傳輸的數據並將之傳送到指定的地方，成功后從channel中刪除。
2>.StreamSets
一.什么是StreamSets 在StreamSets推出前，Flume，Scribe等少數開源工具是流式采集日志僅有的解決方案，Flume的應用案例最多。 二.StreamSets優缺點 StreamSets是Flume的良好替代者，優勢在於： （1）功能上 有管理界面，可以單個流啟停，統計報表豐富，可以預覽數據。因為其源和目標的支持特別豐富，還可以對數據進行不落地處理，因此還可以替代傳統ETL軟件的一部分功能 （2）源端支持 其多出HDFS、JDBC、Redis、FTP等幾種重要的源。 （3）目標端支持 其多出JDBC、Redis、RabbitMQ、Flume等幾種重要的目標。 （4）數據處理上 StreamSets有多種字段處理組件，Flume僅有過濾功能。有強大的格式處理能力，且支持源端壓縮格式。還能使用JavaScript和Jython等自定義處理邏輯。 StreamSets缺點： 資源占用率比Flume略高，但因為和Flume一樣可以分布式部署，問題不大。
3>.Fluentd
一.什么是Fluentd Fluentd是另一個開源的數據收集框架。Fluentd使用 C/Ruby開發，使用JSON文件來統一日志數據。它的可插拔架構，支持各種不同種類和格式的數據源和數據輸出。最后它也同時提供了高可靠和很好的擴展性。 官網地址：https://docs.fluentd.org/ 二.Fluentd組件 Fluentd的Input/Buffer/Output非常類似於Flume的 Source/Channel/Sink。
Input：
負責接收數據或者主動抓取數據。支持 syslog，http，file tail等。
Buffer：
負責數據獲取的性能和可靠性，也有文件或內存等不同類型的Buffer可以配置。
Output：
負責輸出數據到目的地例如文件，AWS S3或 者其它的Fluentd
4>.Logstash
Logstash是著名的開源數據棧ELK (ElasticSearch, Logstash, Kibana) 中的那個L。用JRuby開發，運行時依賴JVM。 Logstash本身功能比較單調。幾乎在大部分的情況下ELK作為一個棧是被同時使用的。當你的數據系統需要采集分析日志時，Logstash是首選。否則不建議用。如果只是單純的日志采集也不推薦使用Logstash，因為它占用資源較大，官方推出的Filebeat（是Elastic Stack推出的Beats實現的一種，官網鏈接：https://www.elastic.co/cn/products/beats）可以輕松實現日志收集。 GitHub地址：https://github.com/elastic/logstash。 官網地址：https://www.elastic.co/cn/products/logstash。
5>.Scribe日志收集工具
Scribe是facebook開源的日志收集系統，在facebook內部已經得到大量的應用。它能夠從各種日志源上收集日志，存儲到一個中央存儲系統（可以是NFS，分布式文件系統等）上，以便於進行集中統計分析處理。它為日志的“分布式收集，統一處理”提供了一個可擴展的，高容錯的方案。當中央存儲系統的網絡或者機器出現故障時，scribe會將日志轉存到本地或者另一個位置，當中央存儲系統恢復后，scribe會將轉存的日志重新傳輸給中央存儲系統。 GitHub地址：https://github.com/facebookarchive/scribe Scribe資料相對較少，網上的案例很少，大多都是理論派偏多，因此推薦大家使用開源社區比較火熱的日志收集工具。如Flume，FileBeats之類的。
6>.chukwa
官方關於Chukwa是這樣介紹的： “Apache Chukwa is an open source data collection system for monitoring large distributed systems. Apache Chukwa is built on top of the Hadoop Distributed File System (HDFS) and Map/Reduce framework and inherits Hadoop’s scalability and robustness. Apache Chukwa also includes a flexible and powerful toolkit for displaying, monitoring and analyzing results to make the best use of the collected data.” 大致翻譯如下： Apache Chukwa是一個用於監控大型分布式系統的開源數據收集系統。Apache Chukwa構建於Hadoop分布式文件系統（HDFS）和Map / Reduce框架之上，並繼承了Hadoop的可擴展性和健壯性。Apache Chukwa還包括一個靈活而強大的工具包，用於顯示，監控和分析結果，以充分利用收集的數據。 官方地址：http://chukwa.apache.org/ 我們在官網可以看到Chukwa最近一次發布時間是2016-10-08,很顯然，該項目的代碼都已經多年未更新了，在國內的案例也非常非常少，不推薦使 用。
三.消息隊列組件
1>.Apache Kafka
一.什么是Kafka
Kafka是一個高吞吐、分布式、基於發布訂閱的消息系統，利用Kafka可在廉價PC server上搭建起大規模的消息系統。 官網地址：http://kafka.apache.org/。
二.Kafka特性： （1）使用zero-copy技術，數據在磁盤上存取代價為O(1)。 （2）高吞吐率，在萬兆網下，單點寫入吞吐率高於300MB/s，這取決於它順序寫入磁盤，其效率緊追隨機寫入內存的速度。 （3）顯式分布式，即所有的producer、broker和consumer都可為分布式的。 （4）消費者的high level API易於使用（low level API則相當坑）
三.Kafka核心概念： （1）Broker：
Kafka集群包含一個或多個服務實例，這種服務實例被稱為broker。 （2）Topic：
每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。 （3）Producer：
負責發布（寫入）消息到Kafka Broker。 （4）Consumer：
消息消費者，向Kafka Broker訂閱（讀取）消息的客戶端。
2>.RabbitMQ
一.什么是RabbitMQ RabbitMQ是使用Erlang語言開發的開源消息隊列系統，基於AMQP協議來實 現。AMQP的主要特征是面向消息、隊列、路由（包括點對點和發布/訂 閱）、可靠性、安全。 官網地址：https://www.rabbitmq.com/ 二.RabbitMQ的使用場景 AMQP協議更多用在企業系統內，對數據一致性、穩定性和可靠性要求很高的場景，對性能和吞吐量的要求還在其次。 因此，RabbitMQ比較適合作為業務系統的消息隊列使用，尤其像金融、通信等領域。在大數據領域，通常還是將吞吐量作為首要考慮因素，Kafka比RabbitMQ更適合。
3>.RocketMQ
一.什么是RocketMQ 如上圖所示（原鏈接：http://rocketmq.apache.org/docs/rmq-arc/），Apache RocketMQ是一個分布式消息傳遞和流媒體平台，具有低延遲，高性能和可靠性，萬億級容量和靈活的可擴展性。它由四部分組成：名稱服務器，代理，生產者和消費者。它們中的每一個都可以水平擴展而沒有單一的故障點。 官網地址：http://rocketmq.apache.org/。 二.RocketMQ的使用場景 RocketMQ可以看作在Kafka的頂層設計上增加了一些電商業務場景支持的 產物，但在性能、吞吐量上相比Kafka並沒有優勢，且與其他大數據組件 的集成不好。如果你不需要如分布式事務、定時消息等這些額外特性，則沒有必要使用RocketMQ。
四.其他組件
1>.Zookeeper
一.什么是zookeeper
ZooKeeper是一個分布式應用程序協調服務，是Google的Chubby的一個開源實現
二.Znode
ZooKeeper名字空間的每個節點都是以這樣一個路徑來標識的。這樣的節點統一稱為znode。 持久的/臨時的 無序的/有序的
三.Zookeeper角色 Leader：接收消息，並編號 Follower：同步消息，參與Leader選舉 Observer：同步消息，但不參與Leader選舉
三.ZAB
ZooKeeper原子廣播協議，是其數據一致性算法，與Paxos有着明顯區別（注意ZooKeeper並不是強一致的）
2>.Storm
一.什么是Storm
Apache Storm是一個分布式實時大數據處理系統。Storm設計用於在容錯和水平可擴展方法中處理大量數據。它是一個流數據框架，具有較高的吞吐率和較低的延遲。Storm是無狀態的，它通過Apache Zookeeper管理分布式環境和集群狀態。部署和開發Storm任務比較簡單，您可以並行地對實時數據執行各種操作。
由於Storm是用Clojure語言開發的，這種語言入門門檻較高，因此想從源碼層學習Storm有較大的難度。阿里使用Java重寫了Strom並做了一些改進，稱JStorm，但目前由於阿里實時計算轉向Flink，JStorm也已不維護了。
官方地址：http://storm.apache.org/
二.Storm核心概念
（1）Tuple
是Storm中的主要數據結構。它是有序元素的列表。默認情況下，Tuple支持所有數據類型。通常，它被建模為一組逗號分隔的值，並傳遞到Storm集群。
（2）Stream
是Tuple的無序序列。
（4）Spouts
流的源。通常Storm從原始數據源（如Twritter Streaming API,Apache Kafa隊列等）接受輸入數據。你也可以編寫spouts以從數據源讀取數據。“ISpout”是實現的核心接口，一些特定的接口是IRichSpout，BaseRichSpout，KafkaSpout等。
（5）Bolts
是邏輯處理單元。Spouts將數據從傳遞到Bolts，經過處理產生新的輸入流。Bolts可以執行過濾，聚合，連接，與數據源和數據庫交互等操作。Bolts接受數據並發射到一個或多個Bolts。“IBolt”是實現Bolts的核心接口。一些常見的接口是IRichBolt，IBasicBolt等。
（6）Topplogy
也稱拓撲，Spouts和Bolts連接在一起，形成拓撲結構。實時應用程序邏輯在拓撲中指定。簡單地說，拓撲是有向圖，其中頂點是計算，邊緣是數據流。拓撲從Spouts開始，Spouts將數據發射到一個或多個Bolts。Bolt表示拓撲中具有最小處理邏輯的節點，並且Bolts的輸出可以發射到另一個Bolts作為輸入。Storm保持拓撲始終運行，直到手動終止拓撲。
（7）Stream groupings
流分組。數據流從Spouts流到Bolts，或從一個Bolt流到另一個Bolt。流分組控制元組在拓撲中的路由方式，並幫助我們了解拓撲中的元組流。在當前版本中有8種流分組（另外也可以自定義）：
Shuffle grouping（隨機）
Fields grouping（按字段分組）
Partial Key grouping（帶負載均衡的按字段分組）
All grouping（復制給所有Bolts的Tasks）
Global grouping（傳遞一個Bolt的Tasks）
None grouping（不關心如果分組）
Direct grouping（由tuple的生成者決定發送給哪個Task）
Local or shuffle grouping（若目標Bolt有多個worker進程，會發送給這些進程的Tasks；否則執行Shuffle grouping）
3>.Spark
一.Spark的產生背景 傳統式的Hadoop缺點主要有以下兩點： 第一.迭代式計算效率低（一個MapReduce依賴上一個MapReduce的結果）; 第二.交互式數據挖掘效率低（運行一個HIVE語句效率是極低的，第一天輸入的SQL可能等到第二天才能拿到結果） Spark優化了Hadoop的兩個缺點，可以將多個job合並成一個job來執行，也可以將於磁盤的交互遷移到內存進行交互，從而提升了工作效率。 二.什么是Spark： Apache Spark是一種快速通用的集群計算系統。它提供Java，Scala，Python和R中的高級API，以及支持通用執行圖的優化引擎。它還支持一組豐富的更高級別的工具，包括星火SQL用於SQL和結構化數據的處理，MLlib機器學習，GraphX用於圖形處理和星火流。 Spark設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計算。為了實現這樣的要求，同時獲得最大靈活性，Spark支持在各種集群管理器(cluster manager)上運行，包括Hadoop YARN、Apache Mesos，以及Spark自帶的一個簡易調度器，叫作獨立調度器。 Spark得到了眾多大數據公司的支持，這些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、騰訊、京東、攜程、優酷土豆。當前百度的Spark已應用於鳳巢、大搜索、直達號、百度大數據等業務；阿里利用GraphX構建了大規模的圖計算和圖挖掘系統，實現了很多生產系統的推薦算法；騰訊Spark集群達到8000台的規模，是當前已知的世界上最大的Spark集群。 三.Spark內置組件 （1）Spark Core： 實現了Spark的基本功能，包含任務調度、內存管理、錯誤恢復、與存儲系統 交互等模塊。Spark Core 中還包含了對彈性分布式數據集(resilient distributed dataset，簡稱RDD)的 API定義。 （2）Spark SQL： 是Spark用來操作結構化數據的程序包。通過 Spark SQL，我們可以使用 SQL 或者 Apache Hive 版本的 SQL 方言(HQL)來查詢數據。Spark SQL 支持多種數據源，比 如 Hive 表、Parquet 以及 JSON 等。 （3）Spark Streaming： 是Spark提供的對實時數據進行流式計算的組件。提供了用來操作數據流的 API，並且與 Spark Core 中的 RDD API 高度對應。 （4）Spark MLlib： 提供常見的機器學習(ML)功能的程序庫。包括分類、回歸、聚類、協同過濾等，還提供了模型評估、數據 導入等額外的支持功能。 四.Spark的安裝模式 如上圖所示，Spark的安裝模式可分為:Local、Local-Cluster、Standalone、Yarn、Mesos Master節點主要運行集群管理器的中心化部分，所承載的作用是分配Application到Worker節點，維護Worker節點，Driver，Application的狀態。Worker節點負責具體的業務運行。
4>.Flink
Apache Flink是一個開源的分布式，高性能，高可用，准確的流處理框架。主要由Java代碼實現。支持實時流（stream）處理和批（batch）處理，批數據只是流數據的一個極限特例。Flink原生支持了迭代計算，內存管理和程序優化。 官網地址：https://flink.apache.org/