簡述大數據實時處理框架

時間 2019-11-16

標籤簡述數據實時處理框架简体版

原文原文鏈接

大屏幕上實時跳動的營業額，背後的技術結構。前端

歡迎來到BigData的世界

現現在，咱們來到了數據時代，數據信息化與咱們的生活與工做息息相關。此篇文章簡述利用大數據框架，實時處理數據的流程與相關框架的介紹，主要包括：java

數據實時處理的概念和意義

數據實時處理能作什麼

數據實時處理架構簡介

數據實時處理代碼演示

數據實時處理的概念和意義

什麼是數據實時處理呢？我我的對數據實時處理的理解爲：數據從生成->實時採集->實時緩存存儲->（準）實時計算->實時落地->實時展現->實時分析。這一個流程線下來，處理數據的速度在秒級甚至毫秒級。node

數據實時處理有什麼意義呢？咱們獲得數據能夠進行數據分析，利用數據統計方法，從錯綜複雜的數據關係中梳理出事物的聯繫，好比發展趨勢、影響因素、因果關係等。甚至創建一些BI，對一些數據的有用信息進行可視化呈現，並造成數據故事。算法

數據實時處理能作什麼

數據的實時計算

何爲數據的實時計算？咱們從數據源端拿到數據，可能不盡如人意，咱們想對獲得的數據進行 ETL 操做、或者進行關聯等等，那麼咱們就會用到數據的實時計算。目前主流的實時計算框架有 spark，storm，flink 等。sql

數據的實時落地

數據的實時落地，意思是將咱們的源數據或者計算好的數據進行實時的存儲。在大數據領域，推薦使用 HDFS，ES 等進行存儲。數據庫

數據的實時展現與分析

咱們拿到了數據，要會用數據的價值。數據的價值體如今數據中相互關聯關係，或與歷史關聯，或能預測將來。咱們實時獲得數據，不只可以利用前端框架進行實時展現，還能夠對其中的一些數據進行算法訓練，預測將來走勢等。apache

example：bootstrap

淘寶雙 11 大屏，每一年的雙 11 是淘寶粉絲瘋狂的日子。馬雲會在雙 11 的當天在阿里總部豎起一面大的電子屏幕，展現淘寶這一天的成績。例如成交額，訪問人數，訂單量，下單量，成交量等等。這個電子大屏的背後，就是用到的咱們所說的數據的實時處理。首先，阿里的服務器遍及全國各地，這些服務器收集PC端、手機端等日誌，上報到服務器，在服務上部署數據採集工具。接下來，因爲數據量龐大，須要作數據的緩存緩衝處理。下一步，對原始日誌進行實時的計算，好比篩選出上面所述的各個指標。最後，經過接口或者其餘形式，進行前端屏幕的實時展現。緩存

數據實時處理架構簡介

接下來是咱們介紹的重點，先放一張數據流程圖：前端框架

數據採集端，選用目前採集數據的主流控件 flume。
數據緩衝緩存，選用分佈式消息隊列 kafka。
數據實時計算，選用 spark 計算引擎。
數據存儲位置，選用分佈式數據存儲 ES。
其餘，指從 ES 中拿到數據後進行可視化展現，數據分析等。

下面將分別簡單的介紹下各個組件：

flume

flume 是一個分佈式的數據收集系統,具備高可靠、高可用、事務管理、失敗重啓、聚合和傳輸等功能。數據處理速度快,徹底能夠用於生產環境。

flume 的核心概念有：event，agent，source，channel，sink

event

flume 的數據流由事件 (event) 貫穿始終。event 是 flume 的基本數據單位，它攜帶日誌數據而且攜帶數據的頭信息，這些 event 由 agent 外部的 source 生成，當 source 捕獲事件後會進行特定的格式化，而後 source 會把事件推入 channel 中。能夠把 channel 看做是一個緩衝區，它將保存事件直到 sink 處理完該事件。sink 負責持久化日誌或者把事件推向另外一個 source。

agent

flume 的核心是 agent。agent 是一個 java 進程,運行在日誌收集端,經過 agent 接收日誌,而後暫存起來,再發送到目的地。每臺機器運行一個 agent。 agent 裏面能夠包含多個 source，channel，sink。

source

source 是數據的收集端，負責將數據捕獲後進行特殊的格式化，將數據封裝到 event 裏，而後將事件推入 channel 中。flume 提供了不少內置的 source，支持 avro,log4j,syslog 等等。若是內置的 source 沒法知足環境的需求，flume 還支持自定義 source。

channel

channel 是鏈接 source 和 sink 的組件，你們能夠將它看作一個數據的緩衝區（數據隊列），它能夠將事件暫存到內存中也能夠持久化到本地磁盤上，直到 sink 處理完該事件。兩個較爲經常使用的 channel，MemoryChannel 和 FileChannel。

sink

sink 從 channel 中取出事件，而後將數據發到別處，能夠向文件系統、數據庫、hadoop、kafka，也能夠是其餘 agent 的 source。

flume 的可靠性與可恢復性

flume 的可靠性：當節點出現故障時，日誌可以被傳送到其餘節點上而不會丟失。Flume 提供了可靠性保障，收到數據首先寫到磁盤上，當數據傳送成功後，再刪除；若是數據發送失敗，能夠從新發送。
flume 的可恢復性：可恢復性是靠 channel。

口述抽象，上兩張官網貼圖:

單個 agent 收集數據流程圖

多個 agent 協做處理數據流程圖

kafka

Kafka 是一個高吞吐量的分佈式發佈-訂閱消息系統。企業中通常使用 kafka 作消息中間件，作緩衝緩存處理。須要 zookeeper 分佈式協調組件管理。

kafka 的設計目標：

提供優秀的消息持久化能力，對 TB 級以上數據也能保證常數時間的訪問性能。
高吞吐率。即便在很是廉價的機器上也能作到每臺機每秒 100000 條消息的傳輸。
支持 kafka server 間的消息分區，及分佈式消費，同時保證每一個 partition 內的消息順序傳輸。
同時支持離線數據處理和實時數據處理。

kafka 核心概念

broker：消息中間件處理結點，一個 kafka 節點就是一個 broker，多個 broker 能夠組成一個 kafka 集羣。
topic：主題，kafka 集羣可以同時負責多個 topic 的分發。
partition：topic 物理上的分組，一個 topic 能夠分爲多個 partition，每一個 partition 是一個有序的隊列。
offset：每一個 partition 都由一系列有序的、不可變的消息組成，這些消息被連續的追加到 partition 中。partition 中的每一個消息都有一個連續的序列號叫作 offset，用於 partition 惟一標識一條消息。
producer：負責發佈消息到 kafka broker。
consumer：消息消費者，向 kafka broker讀取消息的客戶端。
consumer group：每一個 consumer 屬於一個特定的 consumer group。

貼兩張官網圖

prodecer-broker-consumer

分區圖

spark

spark 是一個分佈式的計算框架，是我目前認爲最火的計算框架。

spark，是一種"one stack to rulethem all"的大數據計算框架,指望使用一個技術棧就完美地解決大數據領域的各類計算任務。apache 官方,對 spark 的定義是:通用的大數據快速處理引擎（一「棧」式）。

spark組成

spark core 用於離線計算
spark sql 用於交互式查詢
spark streaming，structed streaming 用於實時流式計算
spark MLlib 用於機器學習
spark GraphX 用於圖計算

spark 特色

速度快：spar k基於內存進行計算(固然也有部分計算基於磁盤,好比 shuffle)。
容易上手開發：spark 的基於 rdd 的計算模型，比 hadoop 的基於 map-reduce 的計算模型要更加易於理解,更加易於上手開發,實現各類複雜功能。
通用性：spark 提供的技術組件,能夠一站式地完成大數據領域的離線批處理、交互式查詢、流式計算、機器學習、圖計算等常見的任務。
與其餘技術的完美集成：例如 hadoop，hdfs、hive、hbase 負責存儲，yarn 負責資源調度，spark 負責大數據計算。
極高的活躍度：spark 目前是 apache 的頂級項目,全世界有大量的優秀工程師是 spark 的 committer，而且世界上不少頂級的 IT 公司都在大規模地使用 spark。

貼個spark架構圖

數據實時處理代碼演示

搭建好各個集羣環境

須要搭建 flume 集羣，kafka 集羣，es 集羣，zookeeper 集羣，因爲本例 spark 是在本地模式運行，因此無需搭建 spark 集羣。

配置好組件之間整合的配置文件

搭建好集羣后，根據集羣組件直接的整合關係，配置好配置文件。其中主要的配置爲 flume 的配置，以下圖：

能夠看到，咱們的 agent 的 source 爲 r1，channel 爲 c1，sink 爲 k1，source 爲我本地 nc 服務，收集日誌時，只須要打開 9999 端口就能夠把日誌收集。channel 選擇爲 memory 內存模式。sink 爲 kafka 的 topic8 主題。

開啓各個集羣進程

開啓 zookeeper 服務。其中 QuorumPeerMain 爲 zookeeper 進程。
開啓 kafka 服務。
開啓 es 服務。
開啓 flume 服務。其中 Application 爲 flume 進程。

建立好 es 對應 table

建立好 es 對應的表，表有三個字段，對應代碼裏面的 case class（代碼隨後貼上）。

代碼以下:

package run

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.Logger
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.elasticsearch.spark.rdd.EsSpark


/** * @author wangjx * 測試kafka數據進行統計 kafka自身維護offset（建議使用自定義維護方式維護偏移量） */
object SparkStreamingAutoOffsetKafka {
  //定義樣例類 與es表對應
  case class people(name:String,country:String,age:Int)

  def main(args: Array[String]): Unit = {
    val logger = Logger.getLogger(this.getClass);
    //spark 配置
    val conf = new SparkConf().setAppName("SparkStreamingAutoOffsetKafka").setMaster("local[2]")
    conf.set("es.index.auto.create","true")
    conf.set("es.nodes","127.0.0.1")
    conf.set("es.port","9200")
    //spark streaming實時計算初始化 定義每10秒一個批次 準實時處理 企業通常都是準實時 好比每隔10秒統計近1分鐘的數據等等
    val ssc = new StreamingContext(conf, Seconds(10))
    val spark = SparkSession.builder()
      .config(conf)
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN");
    //設置kafka參數
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "x:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "exactly-once",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    //kafka主題
    val topic = Set("kafka8")
    //從kafka獲取數據
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topic, kafkaParams)
    )
    //具體的業務邏輯
    val kafkaValue: DStream[String] = stream.flatMap(line=>Some(line.value()))
    val peopleStream = kafkaValue
      .map(_.split(":"))
      //造成people樣例對象
      .map(m=>people(m(0),m(1),m(2).toInt))
    //存入ES
    peopleStream.foreachRDD(rdd =>{
      EsSpark.saveToEs(rdd, "people/man")
    })
    //啓動程序入口
    ssc.start()
    ssc.awaitTermination()
  }
}
複製代碼