流式計算基礎

01_流式計算基礎_第1天（Storm是什麼、Storm核心組件、Storm編程模型）

課程介紹

課程名稱：Storm是什麼

課程目標：

         經過該課程的學習可以瞭解離線計算與流式計算的區別、掌握Storm框架的基礎知識、瞭解流式計算的通常架構圖。

課程大綱：

一、 離線計算是什麼？

二、 流式計算是什麼？

三、 流式計算與離線計算的區別？

四、 Storm是什麼？

五、 Storm與Hadoop的區別？

六、 Storm的應用場景及行業案例

七、 Storm的核心組件（重點掌握）

八、 Storm的編程模型（重點掌握）

九、 流式計算的通常架構圖(重點掌握)

 

 

背景介紹

Storm背景介紹

課程內容

一、離線計算是什麼？

         離線計算：批量獲取數據、批量傳輸數據、週期性批量計算數據、數據展現

         表明技術：Sqoop批量導入數據、HDFS批量存儲數據、MapReduce批量計算數據、Hive批量計算數據、***任務調度

1，hivesql

二、調度平臺

三、Hadoop集羣運維

四、數據清洗（腳本語言）

五、元數據管理

六、數據稽查

七、數據倉庫模型架構

 

二、流式計算是什麼

         流式計算：數據實時產生、數據實時傳輸、數據實時計算、實時展現

         表明技術：Flume實時獲取數據、Kafka/metaq實時數據存儲、Storm/JStorm實時數據計算、Redis實時結果緩存、持久化存儲(mysql)。

         一句話總結：將源源不斷產生的數據實時收集並實時計算，儘量快的獲得計算結果

 

三、離線計算與實時計算的區別

         最大的區別：實時收集、實時計算、實時展現

 

四、Storm是什麼？

         Flume實時採集，低延遲

         Kafka消息隊列，低延遲

         Storm實時計算，低延遲

         Redis實時存儲，低延遲

 

         Storm用來實時處理數據，特色：低延遲、高可用、分佈式、可擴展、數據不丟失。提供簡單容易理解的接口，便於開發。

 

        

         海量數據？數據類型不少，產生數據的終端不少，處理數據能力加強

        

        

五、Storm與Hadoop的區別

l  Storm用於實時計算，Hadoop用於離線計算。

l  Storm處理的數據保存在內存中，源源不斷；Hadoop處理的數據保存在文件系統中，一批一批。

l  Storm的數據經過網絡傳輸進來；Hadoop的數據保存在磁盤中。

l  Storm與Hadoop的編程模型類似

 

 

Job：任務名稱

JobTracker：項目經理

TaskTracker：開發組長、產品經理

Child:負責開發的人員

Mapper/Reduce:開發人員中的兩種角色，一種是服務器開發、一種是客戶端開發

 

Topology:任務名稱

Nimbus:項目經理

Supervisor:開組長、產品經理

Worker:開人員

Spout/Bolt：開人員中的兩種角色，一種是服務器開發、一種是客戶端開發

 

六、Storm應用場景及行業案例

                  Storm用來實時計算源源不斷產生的數據，如同流水線生產。

6.一、運用場景

l  日誌分析

從海量日誌中分析出特定的數據，並將分析的結果存入外部存儲器用來輔佐決策。

l  管道系統

將一個數據從一個系統傳輸到另一個系統，好比將數據庫同步到Hadoop

l  消息轉化器

將接受到的消息按照某種格式進行轉化，存儲到另一個系統如消息中間件

6.二、典型案列

l  一淘-實時分析系統：實時分析用戶的屬性，並反饋給搜索引擎

最初，用戶屬性分析是經過天天在雲梯上定時運行的MR job來完成的。爲了知足實時性的要求，但願可以實時分析用戶的行爲日誌，將最新的用戶屬性反饋給搜索引擎，可以爲用戶展示最貼近其當前需求的結果。

l  攜程-網站性能監控：實時分析系統監控攜程網的網站性能

利用HTML5提供的performance標準得到可用的指標，並記錄日誌。Storm集羣實時分析日誌和入庫。使用DRPC聚合成報表，經過歷史數據對比等判斷規則，觸發預警事件。

l  阿里媽媽-用戶畫像：實時計算用戶的興趣數據

爲了更加精準投放廣告，阿里媽媽後臺計算引擎須要維護每一個用戶的興趣點（理想狀態是，你對什麼感興趣，就向你投放哪類廣告）。用戶興趣主要基於用戶的歷史行爲、用戶的實時查詢、用戶的實時點擊、用戶的地理信息而得，其中實時查詢、實時點擊等用戶行爲都是實時數據。考慮到系統的實時性，阿里媽媽使用Storm維護用戶興趣數據，並在此基礎上進行受衆定向的廣告投放。

        

七、Storm核心組件（重要）

 

 

l  Nimbus：負責資源分配和任務調度。

l  Supervisor：負責接受nimbus分配的任務，啓動和中止屬於本身管理的worker進程。---經過配置文件設置當前supervisor上啓動多少個worker。

l  Worker：運行具體處理組件邏輯的進程。Worker運行的任務類型只有兩種，一種是Spout任務，一種是Bolt任務。

l  Task：worker中每個spout/bolt的線程稱爲一個task. 在storm0.8以後，task再也不與物理線程對應，不一樣spout/bolt的task可能會共享一個物理線程，該線程稱爲executor。

 

八、Storm編程模型（重要）

 

 

l  Topology：Storm中運行的一個實時應用程序的名稱。（拓撲）

l  Spout：在一個topology中獲取源數據流的組件。

一般狀況下spout會從外部數據源中讀取數據，而後轉換爲topology內部的源數據。

l  Bolt：接受數據而後執行處理的組件,用戶能夠在其中執行本身想要的操做。

l  Tuple：一次消息傳遞的基本單元，理解爲一組消息就是一個Tuple。

l  Stream：表示數據的流向。

九、流式計算通常架構圖（重要）

 

 

l  其中flume用來獲取數據。

l  Kafka用來臨時保存數據。

l  Strom用來計算數據。

l  Redis是個內存數據庫，用來保存數據。

 

02_流式計算基礎_第1天（Storm集羣部署、單詞計數、Stream Grouping）

課程介紹

課程名稱：Storm集羣部署及單詞技術

課程目標：

         經過本課程可以掌握Strom集羣搭建、Storm配置文件、Storm源碼管理、Storm編程模型。

課程大綱：

一、 集羣部署的基本流程

二、 集羣部署的基礎環境準備

三、 Storm集羣部署

四、 Storm集羣的經常使用操做命令

五、 Storm集羣的進程及日誌熟悉

六、 Storm源碼下載及目錄熟悉

七、 Storm 單詞計數案列

 

課程內容

一、         集羣部署的基本流程

集羣部署的流程：下載安裝包、解壓安裝包、修改配置文件、分發安裝包、啓動集羣

注意：

    全部的集羣上都須要配置hosts

    vi  /etc/hosts

       192.168.239.128 storm01 zk01 hadoop01

      192.168.239.129 storm02 zk02 hadoop02

       192.168.239.130 storm03 zk03 hadoop03

 

二、         集羣部署的基礎環境準備

安裝前的準備工做（zk集羣已經部署完畢）

l  關閉防火牆

chkconfig iptables off  && setenforce 0

l  建立用戶

groupadd realtime &&　useradd realtime　&& usermod -a -G realtime realtime

l  建立工做目錄並賦權

mkdir /export

mkdir /export/servers

chmod 755 -R /export

l  切換到realtime用戶下

su realtime

三、Storm集羣部署

3.一、下載安裝包

wget    http://124.202.164.6/files/1139000006794ECA/apache.fayea.com/storm/apache-storm-0.9.5/apache-storm-0.9.5.tar.gz

 

3.二、解壓安裝包

tar -zxvf apache-storm-0.9.5.tar.gz -C /export/servers/

cd /export/servers/

ln -s apache-storm-0.9.5 storm

3.三、修改配置文件

         mv /export/servers/storm/conf/storm.yaml /export/servers/storm/conf/storm.yaml.bak

vi /export/servers/storm/conf/storm.yaml

輸入如下內容：

 

 

 

3.四、分發安裝包

scp -r /export/servers/apache-storm-0.9.5 storm02:/export/servers

而後分別在各機器上建立軟鏈接

cd /export/servers/

ln -s apache-storm-0.9.5 storm

3.五、啓動集羣

l  在nimbus.host所屬的機器上啓動 nimbus服務

cd /export/servers/storm/bin/

nohup ./storm nimbus &

l  在nimbus.host所屬的機器上啓動ui服務

cd /export/servers/storm/bin/

nohup ./storm ui &

l  在其它個點擊上啓動supervisor服務

cd /export/servers/storm/bin/

nohup ./storm supervisor &

3.六、查看集羣

訪問nimbus.host:/8080，便可看到storm的ui界面。

 

 

四、Storm經常使用操做命令

有許多簡單且有用的命令能夠用來管理拓撲，它們能夠提交、殺死、禁用、再平衡拓撲。

l  提交任務命令格式：storm jar 【jar路徑】 【拓撲包名.拓撲類名】 【拓撲名稱】

bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount

l  殺死任務命令格式：storm kill 【拓撲名稱】 -w 10（執行kill命令時能夠經過-w [等待秒數]指定拓撲停用之後的等待時間）

storm kill topology-name -w 10

l  停用任務命令格式：storm deactivte  【拓撲名稱】

storm deactivte topology-name

咱們可以掛起或停用運行中的拓撲。當停用拓撲時，全部已分發的元組都會獲得處理，可是spouts的nextTuple方法不會被調用。銷燬一個拓撲，可使用kill命令。它會以一種安全的方式銷燬一個拓撲，首先停用拓撲，在等待拓撲消息的時間段內容許拓撲完成當前的數據流。

l  啓用任務命令格式：storm activate【拓撲名稱】

        storm activate topology-name

l  從新部署任務命令格式：storm rebalance  【拓撲名稱】

        storm rebalance topology-name

        再平衡使你重分配集羣任務。這是個很強大的命令。好比，你向一個運行中的集羣增長了節點。再平衡命令將會停用拓撲，而後在相應超時時間以後重分配工人，並重啓拓撲。

五、Storm集羣的進程及日誌熟悉

5.一、部署成功以後，啓動storm集羣。

         依次啓動集羣的各類角色

5.二、查看nimbus的日誌信息

在nimbus的服務器上

cd /export/servers/storm/logs

tail -100f /export/servers/storm/logs/nimbus.log

5.三、查看ui運行日誌信息

在ui的服務器上，通常和nimbus一個服務器

cd /export/servers/storm/logs

tail -100f /export/servers/storm/logs/ui.log

5.四、查看supervisor運行日誌信息

在supervisor服務上

cd /export/servers/storm/logs

tail -100f /export/servers/storm/logs/supervisor.log

5.五、查看supervisor上worker運行日誌信息

在supervisor服務上

cd /export/servers/storm/logs

tail -100f /export/servers/storm/logs/worker-6702.log

 

 

(該worker正在運行wordcount程序)

 

六、Storm源碼下載及目錄熟悉

6.一、在Storm官方網站上尋找源碼地址

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

 

 

6.二、點擊文字標籤進入github

點擊Apache/storm文字標籤，進入github

          https://github.com/apache/storm

6.三、拷貝storm源碼地址

在網頁右側，拷貝storm源碼地址

    

 

6.四、使用Subversion客戶端下載

 

 

https://github.com/apache/storm/tags/v0.9.5

6.五、Storm源碼目錄分析（重要）

 

 

擴展包中的三個項目，使storm能與hbase、hdfs、kafka交互

 

 

 

七、Storm單詞技術案例（重點掌握）

7.一、功能說明

設計一個topology，來實現對文檔裏面的單詞出現的頻率進行統計。

整個topology分爲三個部分：

l  RandomSentenceSpout：數據源，在已知的英文句子中，隨機發送一條句子出去。

l  SplitSentenceBolt：負責將單行文本記錄（句子）切分紅單詞

l  WordCountBolt：負責對單詞的頻率進行累加

7.二、項目主要流程

 

 

7.三、RandomSentenceSpout的實現及生命週期

 

 

7.四、SplitSentenceBolt的實現及生命週期

 

 

 

7.五、WordCountBolt的實現及生命週期

 

 

7.六、Stream Grouping詳解

Storm裏面有7種類型的stream grouping

l  Shuffle Grouping: 隨機分組， 隨機派發stream裏面的tuple，保證每一個bolt接收到的tuple數目大體相同。

l  Fields Grouping：按字段分組，好比按userid來分組，具備一樣userid的tuple會被分到相同的Bolts裏的一個task，而不一樣的userid則會被分配到不一樣的bolts裏的task。

l  All Grouping：廣播發送，對於每個tuple，全部的bolts都會收到。

l  Global Grouping：全局分組， 這個tuple被分配到storm中的一個bolt的其中一個task。再具體一點就是分配給id值最低的那個task。

l  Non Grouping：不分組，這stream grouping個分組的意思是說stream不關心到底誰會收到它的tuple。目前這種分組和Shuffle grouping是同樣的效果， 有一點不一樣的是storm會把這個bolt放到這個bolt的訂閱者同一個線程裏面去執行。

l  Direct Grouping： 直接分組， 這是一種比較特別的分組方法，用這種分組意味着消息的發送者指定由消息接收者的哪一個task處理這個消息。只有被聲明爲Direct Stream的消息流能夠聲明這種分組方法。並且這種消息tuple必須使用emitDirect方法來發射。消息處理者能夠經過TopologyContext來獲取處理它的消息的task的id （OutputCollector.emit方法也會返回task的id）。

l  Local or shuffle grouping：若是目標bolt有一個或者多個task在同一個工做進程中，tuple將會被隨機發生給這些tasks。不然，和普通的Shuffle Grouping行爲一致。

 

03_流式計算基礎_第1天（Kafka集羣安裝部署、Kafka生產者、Kafka消費者）

課程介紹

課程名稱：Storm上游數據源之Kakfa

課程目標：

         經過本課程理解Storm消費的數據來源、理解JMS規範、理解Kafka核心組件、掌握Kakfa生產者API、掌握Kafka消費者API。對流式計算的生態環境有深刻的瞭解，具有流式計算項目架構的能力。

課程大綱：

一、 kafka是什麼？

二、 JMS規範是什麼？

三、 爲何須要消息隊列？

四、 Kafka核心組件

五、 Kafka安裝部署

六、 Kafka生產者Java API

七、 Kafka消費者Java API

 

課程內容

一、Kafka是什麼

在流式計算中，Kafka通常用來緩存數據，Storm經過消費Kafka的數據進行計算。

 

KAFKA + STORM +REDIS

 

l  Apache Kafka是一個開源消息系統，由Scala寫成。是由Apache軟件基金會開發的一個開源消息系統項目。

l  Kafka最初是由LinkedIn開發，並於2011年初開源。2012年10月從Apache Incubator畢業。該項目的目標是爲處理實時數據提供一個統1、高通量、低等待的平臺。

l  Kafka是一個分佈式消息隊列：生產者、消費者的功能。它提供了相似於JMS的特性，可是在設計實現上徹底不一樣，此外它並非JMS規範的實現。

l  Kafka對消息保存時根據Topic進行歸類，發送消息者稱爲Producer,消息接受者稱爲Consumer,此外kafka集羣有多個kafka實例組成，每一個實例(server)成爲broker。

l  不管是kafka集羣，仍是producer和consumer都依賴於zookeeper集羣保存一些meta信息，來保證系統可用性

二、JMS是什麼

2.一、JMS的基礎

         JMS是什麼：JMS是Java提供的一套技術規範

JMS幹什麼用：用來異構系統 集成通訊，緩解系統瓶頸，提升系統的伸縮性加強系統用戶體驗，使得系統模塊化和組件化變得可行並更加靈活

經過什麼方式：生產消費者模式（生產者、服務器、消費者）

 

 

jdk，kafka，activemq……

2.二、JMS消息傳輸模型

l  點對點模式（一對一，消費者主動拉取數據，消息收到後消息清除）

點對點模型一般是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特色是發送到隊列的消息被一個且只有一個接收者接收處理，即便有多個消息監聽者也是如此。

l  發佈/訂閱模式（一對多，數據生產後，推送給全部訂閱者）

發佈訂閱模型則是一個基於推送的消息傳送模型。發佈訂閱模型能夠有多種不一樣的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的全部消息，即時當前訂閱者不可用，處於離線狀態。

 

 

queue.put（object）  數據生產

queue.take(object)    數據消費

2.三、JMS核心組件

l  Destination：消息發送的目的地，也就是前面說的Queue和Topic。

l  Message [m1] ：從字面上就能夠看出是被髮送的消息。

l  Producer： 消息的生產者，要發送一個消息，必須經過這個生產者來發送。

l  MessageConsumer： 與生產者相對應，這是消息的消費者或接收者，經過它來接收一個消息。

 

 

 

經過與ConnectionFactory能夠得到一個connection

經過connection能夠得到一個session會話。

 

2.四、常見的類JMS消息服務器

2.4.一、JMS消息服務器 ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個徹底支持JMS1.1和J2EE 1.4規範的。

主要特色：

l  多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP

l  徹底支持JMS1.1和J2EE 1.4規範 (持久化,XA消息,事務)

l  對Spring的支持,ActiveMQ能夠很容易內嵌到使用Spring的系統裏面去,並且也支持Spring2.0的特性

l  經過了常見J2EE服務器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中經過JCA 1.5 resource adaptors的配置,可讓ActiveMQ能夠自動的部署到任何兼容J2EE 1.4 商業服務器上

l  支持多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

l  支持經過JDBC和journal提供高速的消息持久化

l  從設計上保證了高性能的集羣,客戶端-服務器,點對點

l  支持Ajax

l  支持與Axis的整合

l  能夠很容易得調用內嵌JMS provider,進行測試

2.4.二、分佈式消息中間件 Metamorphosis

Metamorphosis (MetaQ) 是一個高性能、高可用、可擴展的分佈式消息中間件，相似於LinkedIn的Kafka，具備消息存儲順序寫、吞吐量大和支持本地和XA事務等特性，適用於大吞吐量、順序消息、廣播和日誌數據傳輸等場景，在淘寶和支付寶有着普遍的應用，現已開源。

主要特色：

l  生產者、服務器和消費者均可分佈

l  消息存儲順序寫

l  性能極高,吞吐量大

l  支持消息順序

l  支持本地和XA事務

l  客戶端pull，隨機讀,利用sendfile系統調用，zero-copy ,批量拉數據

l  支持消費端事務

l  支持消息廣播模式

l  支持異步發送消息

l  支持http協議

l  支持消息重試和recover

l  數據遷移、擴容對用戶透明

l  消費狀態保存在客戶端

l  支持同步和異步複製兩種HA

l  支持group commit

2.4.三、分佈式消息中間件 RocketMQ

RocketMQ 是一款分佈式、隊列模型的消息中間件，具備如下特色：

l  可以保證嚴格的消息順序

l  提供豐富的消息拉取模式

l  高效的訂閱者水平擴展能力

l  實時的消息訂閱機制

l  億級消息堆積能力

l  Metaq3.0 版本更名，產品名稱改成RocketMQ

2.4.四、其餘MQ

l  .NET消息中間件 DotNetMQ

l  基於HBase的消息隊列 HQueue

l  Go 的 MQ 框架 KiteQ

l  AMQP消息服務器 RabbitMQ

l  MemcacheQ 是一個基於 MemcacheDB 的消息隊列服務器。

 

 

三、爲何須要消息隊列（重要）

消息系統的核心做用就是三點：解耦，異步和並行

以用戶註冊的案列來講明消息系統的做用

3.一、用戶註冊的通常流程

 

 

問題：隨着後端流程愈來愈多，每步流程都須要額外的耗費不少時間，從而會致使用戶更長的等待延遲。

3.二、用戶註冊的並行執行

 

 

問題：系統並行的發起了4個請求，4個請求中，若是某一個環節執行1分鐘，其餘環節再快，用戶也須要等待1分鐘。若是其中一個環節異常以後，整個服務掛掉了。

 

 

3.三、用戶註冊的最終一致

 

 

一、 保證主流程的正常執行、執行成功以後，發送MQ消息出去。

二、 須要這個destination的其餘系統經過消費數據再執行，最終一致。

 

 

四、Kafka核心組件

l  Topic ：消息根據Topic進行歸類

l  Producer：發送消息者

l  Consumer：消息接受者

l  broker：每一個kafka實例(server)

l  Zookeeper：依賴集羣保存meta信息。

 

 

五、Kafka集羣部署

5.1集羣部署的基本流程

下載安裝包、解壓安裝包、修改配置文件、分發安裝包、啓動集羣

5.2集羣部署的基礎環境準備

安裝前的準備工做（zk集羣已經部署完畢）

l  關閉防火牆

chkconfig iptables off  && setenforce 0

l  建立用戶

groupadd realtime &&　useradd realtime　&& usermod -a -G realtime realtime

l  建立工做目錄並賦權

mkdir /export

mkdir /export/servers

chmod 755 -R /export

l  切換到realtime用戶下

su realtime

5.3 Kafka集羣部署

5.3.一、下載安裝包

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

在linux中使用wget命令下載安裝包

        wget http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz

5.3.二、解壓安裝包

tar -zxvf /export/software/kafka_2.11-0.8.2.2.tgz -C /export/servers/

cd /export/servers/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.三、修改配置文件

cp   /export/servers/kafka/config/server.properties

/export/servers/kafka/config/server.properties.bak

vi  /export/servers/kafka/config/server.properties

輸入如下內容：

 

 

5.3.四、分發安裝包

scp -r /export/servers/kafka_2.11-0.8.2.2 kafka02:/export/servers

而後分別在各機器上建立軟連

cd /export/servers/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.五、再次修改配置文件（重要）

依次修改各服務器上配置文件的的broker.id，分別是0,1,2不得重複。

5.3.六、啓動集羣

依次在各節點上啓動kafka

bin/kafka-server-start.sh  config/server.properties

 

5.四、Kafka經常使用操做命令

l  查看當前服務器中的全部topic

bin/kafka-topics.sh --list --zookeeper  zk01:2181

l  建立topic

./kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 1 --partitions 3 --topic first

l  刪除topic

sh bin/kafka-topics.sh --delete --zookeeper zk01:2181 --topic test

須要server.properties中設置delete.topic.enable=true不然只是標記刪除或者直接重啓。

l  經過shell命令發送消息

kafka-console-producer.sh --broker-list kafka01:9092 --topic itheima

l  經過shell消費消息

sh bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic test1

l  查看消費位置

sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zk01:2181 --group testGroup

l  查看某個Topic的詳情

sh kafka-topics.sh --topic test --describe --zookeeper zk01:2181

 

六、Kafka生產者Java API

 

 

七、Kafka消費者Java API

 

 

---

 

StreamMessage：Java 數據流消息，用標準流操做來順序的填充和讀取。

MapMessage：一個Map類型的消息；名稱爲 string 類型，而值爲 Java 的基本類型。

TextMessage：普通字符串消息，包含一個String。

ObjectMessage：對象消息，包含一個可序列化的Java 對象

BytesMessage：二進制數組消息，包含一個byte[]。

XMLMessage:  一個XML類型的消息。

最經常使用的是TextMessage和ObjectMessage。

 

 04_流式計算基礎_第1天（流式計算案列-實時業務數據計算）

課程介紹

課程名稱：

         Storm實時交易金額計算

課程目標：

         經過本模塊的學習，可以掌握流式計算的基本開發流程，將Kafka+Storm+Redis三門技術集成運用；掌握如何根據業務需求開發一個Storm程序。

課程大綱：

一、 業務背景介紹

二、 業務需求分析

三、 架構設計

四、 功能分析之數據準備

五、 功能分析之數據計算

六、 功能分析之數據展現

七、 工程設計

八、 代碼開發

九、 結果展現

 

課程內容

一、業務背景

根據訂單mq，快速計算雙11當天的訂單量、銷售金額。

 

 

 

 

 

 

二、架構設計及思路

支付系統+kafka+storm/Jstorm集羣+redis集羣

一、支付系統發送mq到kafka集羣中，編寫storm程序消費kafka的數據並計算實時的訂單數量、訂單數量

二、將計算的實時結果保存在redis中

三、外部程序訪問redis進羣中的數據實時展現結果

        

三、數據準備

 

 

訂單編號、訂單時間、支付編號、支付時間、商品編號、商家名稱、商品價格、優惠價格、支付金額

 

 

 

四、業務口徑

l  訂單總數：一條支付信息當一條訂單處理，假設訂單信息不會重發（實際狀況要考慮訂單去重的狀況，父子訂單等多種狀況），計算接收到MQ的總條數，即當作訂單數。

l  銷售額：累加全部的訂單中商品的價格

l  支付金額：累加全部訂單中商品的支付價格

l  用戶人數：一條支付信息當一我的處理，假設訂單一我的只下一單（實際狀況要考慮用戶去重的狀況）。

總體淘寶的業務指標，每一個品類，每一個產品線，每一個淘寶店

 

五、數據展現

讀取redis中的數據，每秒進行展現，打印在控制檯。

 

六、工程設計

l  數據產生：編寫kafka數據生產者，模擬訂單系統發送mq

l  數據輸入：使用PaymentSpout消費kafka中的數據

l  數據計算：使用CountBolt對數據進行統計

l  數據存儲：使用Sava2RedisBolt對數據進行存儲，將結果數據存儲到redis中

l  數據展現：編寫java app客戶端，訪問redis，對數據進行展現，展現方式爲打印在控制檯。

 

一、獲取外部數據源，MQSpout----Open(鏈接你的RMQ)---nextTuple()-----emit（json）

二、ParserPaymentInfoBolt()----execute(Tuple)------解析Json----JavaBean

   productId,orderId,time,price（原價，訂單價，優惠價，支付價），user，收貨地址

   total：原價、total：訂單價、total：訂單人數……

三、Save2ReidsBolt，保存相關業務指標

       問題：   在redis中存放整個網站銷售的原價，  b:t:p:20160410 ---> value

                  redis:   String----> value1+value2 + value3 + value4  incrBy

 

b:t:p:20160410

b:t:p:20161111

b:t:p:20160412

 

七、代碼開發

 

05_流式計算基礎_第2天（Storm目錄樹、任務提交、消息容錯）

課程介紹

課程名稱：

         Storm技術加強

         注：學習本課程，請先學習Storm基礎

課程目標：

         經過本模塊的學習，可以掌握Storm底層的通訊機制、消息容錯機制、storm目錄樹及任務提交流程。

課程大綱：

一、 Storm程序的併發機制

二、 Storm框架通訊機制（worker內部通訊與外部通訊）

三、 Storm組件本地目錄樹

四、 Storm zookeeper目錄樹

五、 Storm 任務提交的過程

課程內容

一、Storm程序的併發機制

1.一、概念

l  Workers (JVMs): 在一個物理節點上能夠運行一個或多個獨立的JVM 進程。一個Topology能夠包含一個或多個worker(並行的跑在不一樣的物理機上), 因此worker process就是執行一個topology的子集, 而且worker只能對應於一個topology 

l  Executors (threads): 在一個worker JVM進程中運行着多個Java線程。一個executor線程能夠執行一個或多個tasks。但通常默認每一個executor只執行一個task。一個worker能夠包含一個或多個executor, 每一個component (spout或bolt)至少對應於一個executor, 因此能夠說executor執行一個compenent的子集, 同時一個executor只能對應於一個component。 

l  Tasks(bolt/spout instances)：Task就是具體的處理邏輯對象，每個Spout和Bolt會被看成不少task在整個集羣裏面執行。每個task對應到一個線程，而stream grouping則是定義怎麼從一堆task發射tuple到另一堆task。你能夠調用TopologyBuilder.setSpout和TopBuilder.setBolt來設置並行度 — 也就是有多少個task。 

1.二、配置並行度

l  對於併發度的配置, 在storm裏面能夠在多個地方進行配置, 優先級爲：

defaults.yaml < storm.yaml < topology-specific configuration

< internal component-specific configuration < external component-specific configuration 

l  worker processes的數目, 能夠經過配置文件和代碼中配置, worker就是執行進程, 因此考慮併發的效果, 數目至少應該大亍machines的數目 

l  executor的數目, component的併發線程數，只能在代碼中配置(經過setBolt和setSpout的參數), 例如, setBolt("green-bolt", new GreenBolt(), 2) 

l  tasks的數目, 能夠不配置, 默認和executor1:1, 也能夠經過setNumTasks()配置 

Topology的worker數經過config設置，即執行該topology的worker（java）進程數。它能夠經過 storm rebalance 命令任意調整。 

Config conf = newConfig(); conf.setNumWorkers(2); //用2個worker topologyBuilder.setSpout("blue-spout", newBlueSpout(), 2); //設置2個併發度 topologyBuilder.setBolt("green-bolt", newGreenBolt(), 2).setNumTasks(4).shuffleGrouping("blue-spout"); //設置2個併發度，4個任務 topologyBuilder.setBolt("yellow-bolt", newYellowBolt(), 6).shuffleGrouping("green-bolt"); //設置6個併發度 StormSubmitter.submitTopology("mytopology", conf, topologyBuilder.createTopology());

 

 

 

 

 

3個組件的併發度加起來是10，就是說拓撲一共有10個executor，一共有2個worker，每一個worker產生10 / 2 = 5條線程。

綠色的bolt配置成2個executor和4個task。爲此每一個executor爲這個bolt運行2個task。

 

l  動態的改變並行度

Storm支持在不 restart topology 的狀況下, 動態的改變(增減) worker processes 的數目和 executors 的數目, 稱爲rebalancing. 經過Storm web UI，或者經過storm rebalance命令實現： 

storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10

 

 

 

二、Storm通訊機制

Worker間的通訊常常須要經過網絡跨節點進行，Storm使用ZeroMQ或Netty(0.9之後默認使用)做爲進程間通訊的消息框架。

Worker進程內部通訊：不一樣worker的thread通訊使用LMAX Disruptor來完成。

         不一樣topologey之間的通訊，Storm不負責，須要本身想辦法實現，例如使用kafka等；

2.一、Worker進程間通訊

worker進程間消息傳遞機制，消息的接收和處理的大概流程見下圖

 

 

ü  對於worker進程來講，爲了管理流入和傳出的消息，每一個worker進程有一個獨立的接收線程[m1] (對配置的TCP端口supervisor.slots.ports進行監聽);

對應Worker接收線程，每一個worker存在一個獨立的發送線程[m2] ，它負責從worker的transfer-queue[m3] 中讀取消息，並經過網絡發送給其餘worker

ü  每一個executor有本身的incoming-queue[m4] 和outgoing-queue[m5] 。

Worker接收線程將收到的消息經過task編號傳遞給對應的executor(一個或多個)的incoming-queues;

每一個executor有單獨的線程分別來處理spout/bolt的業務邏輯，業務邏輯輸出的中間數據會存放在outgoing-queue中，當executor的outgoing-queue中的tuple達到必定的閥值，executor的發送線程將批量獲取outgoing-queue中的tuple,併發送到transfer-queue中。

ü  每一個worker進程控制一個或多個executor線程，用戶可在代碼中進行配置。其實就是咱們在代碼中設置的併發度個數。

2.二、Worker進程間通訊分析

 

 

一、 Worker接受線程經過網絡接受數據，並根據Tuple中包含的taskId，匹配到對應的executor；而後根據executor找到對應的incoming-queue，將數據存發送到incoming-queue隊列中。

二、 業務邏輯執行現成消費incoming-queue的數據，經過調用Bolt的execute(xxxx)方法，將Tuple做爲參數傳輸給用戶自定義的方法

三、 業務邏輯執行完畢以後，將計算的中間數據發送給outgoing-queue隊列，當outgoing-queue中的tuple達到必定的閥值，executor的發送線程將批量獲取outgoing-queue中的tuple,併發送到Worker的transfer-queue中

四、 Worker發送線程消費transfer-queue中數據，計算Tuple的目的地，鏈接不一樣的node+port將數據經過網絡傳輸的方式傳送給另外一個的Worker。

五、 另外一個worker執行以上步驟1的操做。

2.三、Worker進程間技術(Netty、ZeroMQ)

2.3.一、Netty

Netty是一個NIO client-server(客戶端服務器)框架，使用Netty能夠快速開發網絡應用，例如服務器和客戶端協議。Netty提供了一種新的方式來使開發網絡應用程序，這種新的方式使得它很容易使用和有很強的擴展性。Netty的內部實現時很複雜的，可是Netty提供了簡單易用的api從網絡處理代碼中解耦業務邏輯。Netty是徹底基於NIO實現的，因此整個Netty都是異步的。

書籍：Netty權威指南

2.3.二、ZeroMQ

ZeroMQ是一種基於消息隊列的多線程網絡庫，其對套接字類型、鏈接處理、幀、甚至路由的底層細節進行抽象，提供跨越多種傳輸協議的套接字。ZeroMQ是網絡通訊中新的一層，介於應用層和傳輸層之間（按照TCP/IP劃分），其是一個可伸縮層，可並行運行，分散在分佈式系統間。

ZeroMQ定位爲：一個簡單好用的傳輸層，像框架同樣的一個socket library，他使得Socket編程更加簡單、簡潔和性能更高。是一個消息處理隊列庫，可在多個線程、內核和主機盒之間彈性伸縮。ZMQ的明確目標是「成爲標準網絡協議棧的一部分，以後進入Linux內核」。

2.四、Worker 內部通訊技術(Disruptor)

2.4.一、Disruptor的來歷

ü  一個公司的業務與技術的關係，通常能夠分爲三個階段。第一個階段就是跟着業務跑。第二個階段是經歷了幾年的時間，才達到的驅動業務階段。第三個階段，技術引領業務的發展乃至企業的發展。因此咱們在學習Disruptor這個技術時，不得不提LMAX這個機構，由於Disruptor這門技術就是由LMAX公司開發並開源的。

ü  LMAX是在英國註冊並受到FSA監管（監管號碼爲509778）的外匯黃金交易所。LMAX也是歐洲第一家也是惟一一家採用多邊交易設施Multilateral Trading Facility（MTF）擁有交易所牌照和經紀商牌照的歐洲頂級金融公司

ü  LAMX擁有最迅捷的交易平臺，頂級技術支持。LMAX交易所使用「（MTF）分裂器Disruptor」技術，能夠在極短期內（通常在3百萬秒之一內）處理訂單，在一個線程裏每秒處理6百萬訂單。全部訂單均爲撮合成交形式，無一例外。多邊交易設施（MTF）曾經用來設計倫敦證券交易 所（london Stock Exchange）、德國證券及衍生工具交易所（Deutsche Borse）和歐洲證券交易所（Euronext）。

ü  2011年LMAX憑藉該技術得到了金融行業技術評選大賽的最佳交易系統獎和甲骨文「公爵杯」創新編程框架獎。

2.4.二、Disruptor是什麼

一、   簡單理解：Disruptor是一個Queue。Disruptor是實現了「隊列」的功能，並且是一個有界隊列。而隊列的應用場景天然就是「生產者-消費者」模型。

二、   在JDK中Queue有不少實現類，包括不限於ArrayBlockingQueue、LinkBlockingQueue，這兩個底層的數據結構分別是數組和鏈表。數組查詢快，鏈表增刪快，可以適應大多數應用場景。

三、   可是ArrayBlockingQueue、LinkBlockingQueue都是線程安全的。涉及到線程安全，就會有synchronized、lock等關鍵字，這就意味着CPU會打架。

四、   Disruptor一種線程之間信息無鎖的交換方式（使用CAS（Compare And Swap/Set）操做）。

2.4.二、Disruptor主要特色

一、   沒有競爭=沒有鎖=很是快。

二、   全部訪問者都記錄本身的序號的實現方式，容許多個生產者與多個消費者共享相同的數據結構。

三、   在每一個對象中都能跟蹤序列號（ring buffer，claim Strategy，生產者和消費者），加上神奇的cache line padding，就意味着沒有爲僞共享和非預期的競爭。

2.4.二、Disruptor 核心技術點

Disruptor能夠當作一個事件監聽或消息機制，在隊列中一邊生產者放入消息，另一邊消費者並行取出處理.

底層是單個數據結構：一個ring buffer。

每一個生產者和消費者都有一個次序計算器，以顯示當前緩衝工做方式。

每一個生產者消費者可以操做本身的次序計數器的可以讀取對方的計數器，生產者可以讀取消費者的計算器確保其在沒有鎖的狀況下是可寫的。

 

核心組件

ü  Ring Buffer 環形的緩衝區，負責對經過 Disruptor 進行交換的數據（事件）進行存儲和更新。

ü  Sequence 經過順序遞增的序號來編號管理經過其進行交換的數據（事件），對數據(事件)的處理過程老是沿着序號逐個遞增處理。

ü  RingBuffer底層是個數組，次序計算器是一個64bit long 整數型，平滑增加。

 

 

 

 

一、 接受數據並寫入到腳標31的位置，以後會沿着序號一直寫入，可是不會繞過消費者所在的腳標。

二、 Joumaler和replicator同時讀到24的位置，他們能夠批量讀取數據到30

三、消費邏輯線程讀到了14的位置，可是無法繼續讀下去，由於他的sequence暫停在15的位置上，須要等到他的sequence給他序號。若是sequence能正常工做，就能讀取到30的數據。

 

 

三、Storm組件本地目錄樹

 

 

四、Storm zookeeper目錄樹

 

 

五、Storm 任務提交的過程

 

 

TopologyMetricsRunnable.TaskStartEvent[oldAssignment=<null>,newAssignment=Assignment[masterCodeDir=C:\Users\MAOXIA~1\AppData\Local\Temp\\e73862a8-f7e7-41f3-883d-af494618bc9f\nimbus\stormdist\double11-1-1458909887,nodeHost={61ce10a7-1e78-4c47-9fb3-c21f43a331ba=192.168.1.106},taskStartTimeSecs={1=1458909910, 2=1458909910, 3=1458909910, 4=1458909910, 5=1458909910, 6=1458909910, 7=1458909910, 8=1458909910},workers=[ResourceWorkerSlot[hostname=192.168.1.106,memSize=0,cpu=0,tasks=[1, 2, 3, 4, 5, 6, 7, 8],jvm=<null>,nodeId=61ce10a7-1e78-4c47-9fb3-c21f43a331ba,port=6900]],timeStamp=1458909910633,type=Assign],task2Component=<null>,clusterName=<null>,topologyId=double11-1-1458909887,timestamp=0]

 

 

 

六、Storm 消息容錯機制

6.一、整體介紹

l  在storm中，可靠的信息處理機制是從spout開始的。

l  一個提供了可靠的處理機制的spout須要記錄他發射出去的tuple，當下遊bolt處理tuple或者子tuple失敗時spout可以從新發射。

l  Storm經過調用Spout的nextTuple()發送一個tuple。爲實現可靠的消息處理，首先要給每一個發出的tuple帶上惟一的ID，而且將ID做爲參數傳遞給SoputOutputCollector的emit()方法：collector.emit(new Values("value1","value2"), msgId); messageid就是用來標示惟一的tupke的，而rootid是隨機生成的

給每一個tuple指定ID告訴Storm系統，不管處理成功仍是失敗，spout都要接收tuple樹上全部節點返回的通知。若是處理成功，spout的ack()方法將會對編號是msgId的消息應答確認；若是處理失敗或者超時，會調用fail()方法。

 

6.二、基本實現

Storm 系統中有一組叫作"acker"的特殊的任務，它們負責跟蹤DAG（有向無環圖）中的每一個消息。

acker任務保存了spout id到一對值的映射。第一個值就是spout的任務id，經過這個id，acker就知道消息處理完成時該通知哪一個spout任務。第二個值是一個64bit的數字，咱們稱之爲"ack val"， 它是樹中全部消息的隨機id的異或計算結果。

 

ack val表示了整棵樹的的狀態，不管這棵樹多大，只須要這個固定大小的數字就能夠跟蹤整棵樹。當消息被建立和被應答的時候都會有相同的消息id發送過來作異或。 每當acker發現一棵樹的ack val值爲0的時候，它就知道這棵樹已經被徹底處理了

 

 

 

 

 

 

 

6.三、可靠性配置

有三種方法能夠去掉消息的可靠性：

將參數Config.TOPOLOGY_ACKERS設置爲0，經過此方法，當Spout發送一個消息的時候，它的ack方法將馬上被調用；

Spout發送一個消息時，不指定此消息的messageID。當須要關閉特定消息可靠性的時候，可使用此方法；

最後，若是你不在乎某個消息派生出來的子孫消息的可靠性，則此消息派生出來的子消息在發送時不要作錨定，即在emit方法中不指定輸入消息。由於這些子孫消息沒有被錨定在任何tuple tree中，所以他們的失敗不會引發任何spout從新發送消息。

 

---

一個worker進程運行一個專用的接收線程來負責將外部發送過來的消息移動到對應的executor線程的incoming-queue中

transfer-queue的大小由參數topology.transfer.buffer.size來設置。transfer-queue的每一個元素實際上表明一個tuple的集合

transfer-queue的大小由參數topology.transfer.buffer.size來設置。

executor的incoming-queue的大小用戶能夠自定義配置。

executor的outgoing-queue的大小用戶能夠自定義配置

 

06_流式計算基礎_第2天（Kafka負載均衡、Kafka自定義Partition、Kafk文件存儲機制）

課程介紹

課程名稱：

         Kafka技術加強

         注：學習本課程請先學習Kafka基礎

課程目標：

    經過本模塊的學習，可以掌握Kafka的負載均衡、Producer生產數據、Kafka文件存儲機制、Kafka自定義partition

課程大綱：

一、 Kafka總體結構圖

二、 Consumer與topic關係

三、 Kafka  Producer消息分發

四、 Consumer 的負載均衡

五、 Kafka文件存儲機制

課程內容

一、Kafka總體結構圖

Kafka名詞解釋和工做方式

 

l  Producer ：消息生產者，就是向kafka broker發消息的客戶端。

l  Consumer ：消息消費者，向kafka broker取消息的客戶端

l  Topic ：咋們能夠理解爲一個隊列。

l  Consumer Group （CG）：這是kafka用來實現一個topic消息的廣播（發給全部的consumer）和單播（發給任意一個consumer）的手段。一個topic能夠有多個CG。topic的消息會複製（不是真的複製，是概念上的）到全部的CG，但每一個partion只會把消息發給該CG中的一個consumer。若是須要實現廣播，只要每一個consumer有一個獨立的CG就能夠了。要實現單播只要全部的consumer在同一個CG。用CG還能夠將consumer進行自由的分組而不須要屢次發送消息到不一樣的topic。

l  Broker ：一臺kafka服務器就是一個broker。一個集羣由多個broker組成。一個broker能夠容納多個topic。

l  Partition：爲了實現擴展性，一個很是大的topic能夠分佈到多個broker（即服務器）上，一個topic能夠分爲多個partition，每一個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的總體（多個partition間）的順序。

l  Offset：kafka的存儲文件都是按照offset.kafka來命名，用offset作名字的好處是方便查找。例如你想找位於2049的位置，只要找到2048.kafka的文件便可。固然the first offset就是00000000000.kafka

二、Consumer與topic關係

本質上kafka只支持Topic；

l  每一個group中能夠有多個consumer，每一個consumer屬於一個consumer group；

一般狀況下，一個group中會包含多個consumer，這樣不只能夠提升topic中消息的併發消費能力，並且還能提升"故障容錯"性，若是group中的某個consumer失效那麼其消費的partitions將會有其餘consumer自動接管。

l  對於Topic中的一條特定的消息，只會被訂閱此Topic的每一個group中的其中一個consumer消費，此消息不會發送給一個group的多個consumer；

那麼一個group中全部的consumer將會交錯的消費整個Topic，每一個group中consumer消息消費互相獨立，咱們能夠認爲一個group是一個"訂閱"者。

l  在kafka中,一個partition中的消息只會被group中的一個consumer消費(同一時刻)；

一個Topic中的每一個partions，只會被一個"訂閱者"中的一個consumer消費，不過一個consumer能夠同時消費多個partitions中的消息。

l  kafka的設計原理決定,對於一個topic，同一個group中不能有多於partitions個數的consumer同時消費，不然將意味着某些consumer將沒法獲得消息。

kafka只能保證一個partition中的消息被某個consumer消費時是順序的；事實上，從Topic角度來講,當有多個partitions時,消息仍不是全局有序的。

三、Kafka消息的分發

         Producer客戶端負責消息的分發

l  kafka集羣中的任何一個broker均可以向producer提供metadata信息,這些metadata中包含"集羣中存活的servers列表"/"partitions leader列表"等信息；

l  當producer獲取到metadata信息以後, producer將會和Topic下全部partition leader保持socket鏈接；

l  消息由producer直接經過socket發送到broker，中間不會通過任何"路由層"，事實上，消息被路由到哪一個partition上由producer客戶端決定；

好比能夠採用"random""key-hash""輪詢"等,若是一個topic中有多個partitions,那麼在producer端實現"消息均衡分發"是必要的。

l  在producer端的配置文件中,開發者能夠指定partition路由的方式。

 

Producer消息發送的應答機制

設置發送數據是否須要服務端的反饋,有三個值0,1,-1

0: producer不會等待broker發送ack

1: 當leader接收到消息以後發送ack

-1: 當全部的follower都同步消息成功後發送ack

         request.required.acks=0

4、Consumer的負載均衡

當一個group中,有consumer加入或者離開時,會觸發partitions均衡.均衡的最終目的,是提高topic的併發消費能力，步驟以下：

一、   假如topic1,具備以下partitions: P0,P1,P2,P3

二、   加入group中,有以下consumer: C1,C2

三、   首先根據partition索引號對partitions排序: P0,P1,P2,P3

四、   根據consumer.id排序: C0,C1

五、   計算倍數: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)

六、   而後依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

 

 

五、kafka文件存儲機制

5.一、Kafka文件存儲基本結構

l  在Kafka文件存儲中，同一個topic下有多個不一樣partition，每一個partition爲一個目錄，partiton命名規則爲topic名稱+有序序號，第一個partiton序號從0開始，序號最大值爲partitions數量減1。

l  每一個partion(目錄)至關於一個巨型文件被平均分配到多個大小相等segment(段)數據文件中。但每一個段segment file消息數量不必定相等，這種特性方便old segment file快速被刪除。默認保留7天的數據。

 

 

l  每一個partiton只須要支持順序讀寫就好了，segment文件生命週期由服務端配置參數決定。（何時建立，何時刪除）

 

 

數據有序的討論？

         一個partition的數據是不是有序的？       間隔性有序，不連續

         針對一個topic裏面的數據，只能作到partition內部有序，不能作到全局有序。

         特別加入消費者的場景後，如何保證消費者消費的數據全局有序的？僞命題。

 

只有一種狀況下才能保證全局有序？就是隻有一個partition。

5.二、Kafka Partition Segment

l  Segment file組成：由2大部分組成，分別爲index file和data file，此2個文件一一對應，成對出現，後綴".index"和「.log」分別表示爲segment索引文件、數據文件。

 

 

l  Segment文件命名規則：partion全局的第一個segment從0開始，後續每一個segment文件名爲上一個segment文件最後一條消息的offset值。數值最大爲64位long大小，19位數字字符長度，沒有數字用0填充。

l  索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。

 

 

3，497：當前log文件中的第幾條信息，存放在磁盤上的那個地方

 

上述圖中索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。

其中以索引文件中元數據3,497爲例，依次在數據文件中表示第3個message(在全局partiton表示第368772個message)、以及該消息的物理偏移地址爲497。

 

l  segment data file由許多message組成， qq物理結構以下：

關鍵字	解釋說明
8 byte offset	在parition(分區)內的每條消息都有一個有序的id號，這個id號被稱爲偏移(offset),它能夠惟一肯定每條消息在parition(分區)內的位置。即offset表示partiion的第多少message
4 byte message size	message大小
4 byte CRC32	用crc32校驗message
1 byte 「magic"	表示本次發佈Kafka服務程序協議版本號
1 byte 「attributes"	表示爲獨立版本、或標識壓縮類型、或編碼類型。
4 byte key length	表示key的長度,當key爲-1時，K byte key字段不填
K byte key	可選
value bytes payload	表示實際消息數據。

 

 

5.三、Kafka 查找message

讀取offset=368776的message，須要經過下面2個步驟查找。

        

 

5.3.一、查找segment file

00000000000000000000.index表示最開始的文件，起始偏移量(offset)爲0

00000000000000368769.index的消息量起始偏移量爲368770 = 368769 + 1

00000000000000737337.index的起始偏移量爲737338=737337 + 1

其餘後續文件依次類推。

以起始偏移量命名並排序這些文件，只要根據offset **二分查找**文件列表，就能夠快速定位到具體文件。當offset=368776時定位到00000000000000368769.index和對應log文件。

5.3.二、經過segment file查找message 

當offset=368776時，依次定位到00000000000000368769.index的元數據物理位置和00000000000000368769.log的物理偏移地址

而後再經過00000000000000368769.log順序查找直到offset=368776爲止。

六、Kafka自定義Partition

         見代碼

 

07_流式計算基礎_第2天（Redis基礎、應用場景、數據結構及案例）

課程介紹

課程名稱：

         Kafka技術加強

         注：學習本課程請先學習Kafka基礎

課程目標：

         經過本模塊的學習，可以掌握Storm底層的通訊機制、消息容錯機制、storm目錄樹及任務提交流程。

課程大綱：

課程內容

一、Redis概述

l  Redis是一個開源，先進的key-value存儲，並用於構建高性能，可擴展的應用程序的完美解決方案。

l  Redis從它的許多競爭繼承來的三個主要特色：

l  Redis數據庫徹底在內存中，使用磁盤僅用於持久性。

l  相比許多鍵值數據存儲，Redis擁有一套較爲豐富的數據類型。String，List,set,map,sortSet

l  Redis能夠將數據複製到任意數量的從服務器。

二、Redis 優點

l  異常快速：Redis的速度很是快，每秒能執行約11萬集合，每秒約81000+條記錄。

l  支持豐富的數據類型：Redis支持字符串、列表、集合、有序集合散列數據類型，這使得它很是容易解決各類各樣的問題。

l  操做都是原子性：全部Redis操做是原子的，這保證了若是兩個客戶端同時訪問的Redis服務器將得到更新後的值。----計數器

l  多功能實用工具：Redis是一個多實用的工具，能夠在多個用例如緩存，消息，隊列使用(Redis原生支持發佈/訂閱)，任何短暫的數據，應用程序，如Web應用程序會話，網頁命中計數等。

三、Redis安裝部署

3.一、Redis下載、編譯、安裝

l  下載redis3.0.5

                  wget http://download.redis.io/releases/redis-3.0.5.tar.gz

l  解壓文件，並建立軟件鏈接

                  tar -zxvf  redis-3.0.5.tar.gz -C /export/servers/

                  ln  –s  redis-3.0.5/  redis

l  編譯redis源碼

                  cd /export/servers/redis

                  make（先安裝gcc）

l  將編譯後的可執行文件安裝到/user/local/redis

                  make PREFIX=/usr/local/redis install

3.二、啓動Redis

l  啓動方式一：Redis前臺默認啓動

進入redis安裝目錄，並啓動Redis服務

                           cd /usr/local/redis/bin/

                           ./redis-server

l  啓動方式二：Redis使用配置文件啓動

拷貝源碼中的redis.conf文件到redis的安裝目錄

                          cp /export/servers/redis/redis.conf        /usr/local/redis/

修改redis.conf的屬性

                          daemonize no  è daemonize yes

啓動redis服務

                           cd  /usr/local/redis

                            bin/redis-server    ./redis.conf

3.三、日誌文件及持久化文件配置

l  修改生成默認日誌文件位置

                  logfile "/usr/local/redis/logs/redis.log"

l  配置持久化文件存放位置

                 dir /usr/local/redis/data/redisData

3.四、Redis客戶端使用

         <dependency>

         <groupId>redis.clients</groupId>

         <artifactId>jedis</artifactId>

         <version>2.8.0</version>

         </dependency>

       

 

 

四、Redis持久化

有兩種持久化方案：RDB和AOF

1） RDB方式按照必定的時間間隔對數據集建立基於時間點的快照。

                  2）AOF方式記錄Server收到的寫操做到日誌文件，在Server重啓時經過回放這些寫操做來重建數據集。該方式相似於MySQL中基於語句格式的binlog。當日志變大時Redis可在後臺重寫日誌。

l  AOF持久化配置

                  1）修改redis.config配置文件，找到appendonly。默認是appendonly no。改爲appendonly yes

                  2）再找到appendfsync 。默認是 appendfsync everysec

                           appendfsync always  

                           #每次收到寫命令就當即強制寫入磁盤，最慢的，可是保證徹底的持久化，不推薦使用  

                           appendfsync everysec    

                            #每秒鐘強制寫入磁盤一次，在性能和持久化方面作了很好的折中，推薦  

                           appendfsync no    

                           #徹底依賴os，性能最好,持久化沒保證  

l  RDB持久化配置

         默認狀況下，Redis保存數據集快照到磁盤，名爲dump.rdb的二進制文件。能夠設置讓Redis在N秒內至少有M次數據集改動時保存數據集，或者你也能夠手動調用SAVE或者BGSAVE命令。 
例如，這個配置會讓Redis在每一個60秒內至少有1000次鍵改動時自動轉儲數據集到磁盤

         save 60 1000 

 

 

五、Redis數據結構

 

 

見代碼