Hadoop2.X管理與開發

Hadoop 2.X 管理與開發

1、Hadoop的起源與背景知識

（一）什麼是大數據  

大數據（Big Data），指沒法在必定時間範圍內用常規軟件工具進行捕捉、管理和處理的數據集合，是須要新處理模式才能具備更強的決策力、洞察發現力和流程優化能力的海量、高增加率和多樣化的信息資產。

大數據的5個特徵（IBM提出）：

l  Volume  （大量）

l  Velocity（高速）

l  Variety （多樣）

l  Value   （價值）

l  Veracity（真實性）

大數據的典型案例：

l  電商網站的商品推薦

l  基於大數據的天氣預報 

（二）OLTP與OLAP

l  OLTP：On-Line Transaction Processing（聯機事務處理過程）。也稱爲面向交易的處理過程，其基本特徵是前臺接收的用戶數據能夠當即傳送到計算中心進行處理，並在很短的時間內給出處理結果，是對用戶操做快速響應的方式之一。OLTP是傳統的關係型數據庫的主要應用，主要是基本的、平常的事務處理，例如銀行交易。

典型案例：銀行轉帳

開啓事務 從轉出帳號中扣錢 往轉入帳號中加錢 提交事務

 

l  OLAP：On-Line Analytic Processing（聯機分析處理過程）。OLAP是數據倉庫系統的主要應用，支持複雜的分析操做，側重決策支持，而且提供直觀易懂的查詢結果。

典型案例：商品推薦

	抽取（讀取）歷史訂單 分析歷史訂單，找到最受歡迎的商品 展現結果

 

 

 

 

 

 

l  OLTP和OLAP的區別：

（三）數據倉庫

數據倉庫，英文名稱爲Data Warehouse，可簡寫爲DW或DWH。數據倉庫，是爲企業全部級別的決策制定過程，提供全部類型數據支持的戰略集合。它是單個數據存儲，出於分析性報告和決策支持目的而建立。

（四）Google的基本思想

Hadoop的思想來源：Google

l  Google搜索引擎，Gmail，安卓，AppspotGoogle Maps，Google earth，Google學術，Google翻譯，Google+，下一步Google what？

l  Google的低成本之道

n  不使用超級計算機，不使用存儲（淘寶的去i，去e，去o之路）

n  大量使用普通的pc服務器（去掉機箱，外設，硬盤），提供有冗餘的集羣服務

n  全世界多個數據中心，有些附帶發電廠

n  運營商向Google倒付費

l  Google的三篇論文（Hadoop的思想來源）

n  GFS（Google File System：Google的文件系統）

n  倒排索引

n  Page Rank（排名前後）

 

n  BigTable（大表）：BigTable是Google設計的分佈式數據存儲系統，用來處理海量的數據的一種非關係型的數據庫。

Eg：什麼是關係模型？（關係型數據庫）

u  常見的NoSQL數據庫（key-value值）：

(1) HBase:     基於HDFS的NoSQL數據庫、面向列的

(2) Redis:      基於內存的NoSQL數據庫、支持持久化: rdb和aof

(3) MongoDB:  面向文檔的NoSQL

(4) Cassandra： 面向列的NoSQL數據庫

l  Hadoop的源起——Lucene，從lucene到nutch，從nutch到hadoop

n  2003-2004年，Google公開了部分GFS和Mapreduce思想的細節，以此爲基礎Doug Cutting等人用了2年業餘時間實現了DFS和Mapreduce機制，使Nutch性能飆升

n  Yahoo招安Doug Cutting及其項目

n  Hadoop於2005 年秋天做爲Lucene的子項目Nutch的 一部分正式引入Apache基金會。2006 年 3 月份，Map-Reduce 和 Nutch Distributed File System (NDFS) 分別被歸入稱爲 Hadoop 的項目中

n  名字來源於Doug Cutting兒子的玩具大象

2、實驗環境

介質	說明
VMware	虛擬機管理器
MySQL	MySQL數據庫的安裝介質
WinSCP-5.9.2-Portable.zip	FTP客戶端工具
Putty	登陸Linux的命令行客戶端
Enterprise-R5-U2-Server-i386-dvd.iso	Oracle Linux安裝介質
jdk-7u75-linux-i586.tar.gz	JDK安裝包
hadoop-2.4.1.tar.gz	Hadoop安裝包
hbase-0.96.2-hadoop2-bin.tar.gz	HBase NoSQL安裝包
hue-3.7.0-cdh5.4.2.tar.gz	HUE（基於Web的管理工具）安裝包
apache-hive-0.13.0-bin.tar.gz	Hive（基於Hadoop的數據倉庫）安裝包
apache-hive-0.13.0-src.tar.gz	Hive（基於Hadoop的數據倉庫）源碼
apache-flume-1.6.0-bin.tar.gz	Flume（日誌採集工具）安裝包
sqoop-1.4.5.bin__hadoop-0.23.tar.gz	Sqoop（HDFS與RDBMS的數據交換）安裝包
pig-0.14.0.tar.gz	Pig（基於Hadoop的數據分析引擎）安裝包
zookeeper-3.4.6.tar.gz	ZooKeeper（實現Hadoop的HA）安裝包

Eg：Hadoop的HDFS和MapReduce示例Demo。

3、Apache Hadoop的體系結構（重點）

（一）分佈式存儲：HDFS

 

l  NameNode（名稱節點）

1．維護HDFS文件系統，是HDFS的主節點。

2．接受客戶端的請求: 上傳文件、下載文件、建立目錄等等

3．記錄客戶端操做的日誌（edits文件），保存了HDFS最新的狀態

①　Edits文件保存了自最後一次檢查點以後全部針對HDFS文件系統的操做，好比：增長文件、重命名文件、刪除目錄等等

②　保存目錄：$HADOOP_HOME/tmp/dfs/name/current

③　可使用hdfs oev -i命令將日誌（二進制）輸出爲XML文件

輸出結果爲：

4．維護文件元信息，將內存中不經常使用（採用LRU算法）的文件元信息保存在硬盤上（fsimage文件）

①　fsimage是HDFS文件系統存於硬盤中的元數據檢查點，裏面記錄了自最後一次檢查點以前HDFS文件系統中全部目錄和文件的序列化信息

②　保存目錄：$HADOOP_HOME/tmp/dfs/name/current

③　可使用hdfs oiv -i命令將日誌（二進制）輸出爲文本（文本和XML）

l  DataNode（數據節點）

1．以數據塊爲單位，保存數據

①　Hadoop 1.0的數據塊大小： 64M

②　Hadoop 2.0 的數據塊大小：128M

2．在全分佈模式下，至少兩個DataNode節點

3．數據保存的目錄：由hadoop.tmp.dir參數指定

例如：

l  Secondary NameNode（第二名稱節點）

1．主要做用是進行日誌合併

2．日誌合併的過程：

l  HDFS存在的問題

①　NameNode單點故障，難以應用二在線場景

解決方案：Hadoop 1.0中，沒有解決方案。

          Hadoop 2.0 中，使用ZooKeeper實現NameNode的HA功能。

②　NameNode壓力過大，且內存受限，影響系統擴展性

解決方案：Hadoop 1.0 中，沒有解決方案。

          Hadoop 2.0 中，使用NameNode的聯盟實現其水平擴展。

（二）YARN: 分佈式計算(MapReduce)

 

l  ResourceManager（資源管理器）

①　接受客戶端的請求: 執行任務

②　分配資源

③　分配任務

l  NodeManager(節點管理器：運行任務MapReduce)

①　從DataNode上獲取數據，執行任務

（三）HBase的體系結構

4、Hadoop 2.X的安裝與配置

（一）Hadoop安裝部署的預備條件

l  安裝Linux

l  安裝JDK

（二）Hadoop的目錄結構

（三）Hadoop安裝部署的三種模式

l  本地模式

l  僞分佈模式

l  全分佈模式

本地模式的配置

參數文件	配置參數	參考值
hadoop-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144

僞分佈模式的配置

參數文件	配置參數	參考值
hadoop-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
hdfs-site.xml	dfs.replication	1
	dfs.permissions	false
core-site.xml	fs.defaultFS	hdfs://<hostname>:9000
	hadoop.tmp.dir	/root/training/hadoop-2.7.3/tmp
mapred-site.xml	mapreduce.framework.name	yarn
yarn-site.xml	yarn.resourcemanager.hostname	<hostname>
	yarn.nodemanager.aux-services	mapreduce_shuffle

全分佈模式的配置

參數文件	配置參數	參考值
hadoop-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
hdfs-site.xml	dfs.replication	2
	dfs.permissions	false
core-site.xml	fs.defaultFS	hdfs://<hostname>:9000
	hadoop.tmp.dir	/root/training/hadoop-2.7.3/tmp
mapred-site.xml	mapreduce.framework.name	yarn
yarn-site.xml	yarn.resourcemanager.hostname	<hostname>
	yarn.nodemanager.aux-services	mapreduce_shuffle
slaves	DataNode的地址	從節點1 從節點2

 

若是出現如下警告信息：

只須要在如下兩個文件中增長下面的環境變量，便可：

l  hadoop-env.sh腳本中

export JAVA_HOME=/root/training/jdk1.8.0_144

export HADOOP_HOME=/root/training/hadoop-2.7.3

export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native

export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib"

l  yarn-env.sh腳本中

export JAVA_HOME=/root/training/jdk1.8.0_144

export HADOOP_HOME=/root/training/hadoop-2.7.3

export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native

export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib"

（四）驗證Hadoop環境

l  HDFS Console：http://192.168.157.11:50070

 

l  Yarn Console：http://192.168.157.11:8088

 

（五）配置SSH免密碼登陸

Demo：配置Linux SSH的免密碼登陸。

5、Hadoop應用案例分析

（一）互聯網應用的架構

①　傳統的架構

 

②　改良後的架構

③　完整的架構圖

（二）日誌分析

①　需求說明

l  對某技術論壇的apache server日誌進行分析，計算論壇關鍵指標，供運營者決策

②　論壇日誌數據有兩部分

l  歷史數據約56GB，統計到2012-05-29

l  自2013-05-30起，天天生成一個數據文件，約150MB

 

③　關鍵指標

④　系統架構

⑤　改良後的系統架構

⑥　HBase表的結構

⑦　日誌分析的執行過程

l  週期性把日誌數據導入到hdfs中

l  週期性把明細日誌導入hbase存儲

l  週期性使用hive進行數據的多維分析

l  週期性把hive分析結果導入到mysql中

（三）Hadoop在淘寶的應用

6、HDFS

（一）HDFS的命令行操做

l  HDFS操做命令（HDFS操做命令幫助信息：hdfs dfs）

命令	說明	示例
-mkdir	在HDFS上建立目錄	l   在HDFS上建立目錄/data      hdfs dfs -mkdir /data l   在HDFS上級聯建立目錄/data/input      hdfs dfs -mkdir -p /data/input
-ls	列出hdfs文件系統根目錄下的目錄和文件	l   查看HDFS根目錄下的文件和目錄      hdfs dfs -ls / l   查看HDFS的/data目錄下的文件和目錄      hdfs dfs -ls /data
-ls -R	列出hdfs文件系統全部的目錄和文件	l   查看HDFS根目錄及其子目錄下的文件和目錄      hdfs dfs -ls -R /
-put	上傳文件或者從鍵盤輸入字符到HDFS	l   將本地Linux的文件data.txt上傳到HDFS      hdfs dfs -put data.txt /data/input l   從鍵盤輸入字符保存到HDFS的文件      hdfs dfs -put - /aaa.txt
-moveFromLocal	與put相相似，命令執行後源文件 local src 被刪除，也能夠從從鍵盤讀取輸入到hdfs file中	hdfs dfs -moveFromLocal data.txt /data/input
-copyFromLocal	與put相相似，也能夠從從鍵盤讀取輸入到hdfs file中	hdfs dfs -copyFromLocal data.txt /data/input
-copyToLocal
-get	將HDFS中的文件被複制到本地	hdfs dfs -get /data/inputdata.txt /root/
-rm	每次能夠刪除多個文件或目錄	hdfs dfs -rm < hdfs file > ...  刪除多個文件 hdfs dfs -rm -r < hdfs dir>... 刪除多個目錄
-getmerge	將hdfs指定目錄下全部文件排序後合併到local指定的文件中，文件不存在時會自動建立，文件存在時會覆蓋裏面的內容	將HDFS上/data/input目錄下的全部文件，合併到本地的a.txt文件中 hdfs dfs -getmerge /data/input /root/a.txt
-cp	拷貝HDFS上的文件
-mv	移動HDFS上的文件
-count	統計hdfs對應路徑下的目錄個數，文件個數，文件總計大小 顯示爲目錄個數，文件個數，文件總計大小，輸入路徑
-du	顯示hdfs對應路徑下每一個文件夾和文件的大小		hdfs dfs -du /
-text、-cat	將文本文件或某些格式的非文本文件經過文本格式輸出
balancer	若是管理員發現某些DataNode保存數據過多，某些DataNode保存數據相對較少，可使用上述命令手動啓動內部的均衡過程

l  HDFS管理命令

    HDFS管理命令幫助信息：hdfs dfsadmin

命令	說明	示例
-report	顯示文件系統的基本數據	hdfs dfsadmin -report
-safemode	HDFS的安全模式命令 < enter | leave | get | wait >	hdfs dfsadmin -safemode enter|leave|get|wait

（二）HDFS的Java API

經過HDFS提供的Java API，咱們能夠完成如下的功能：

①　在HDFS上建立目錄

②　經過FileSystem API讀取數據（下載文件）

③　寫入數據（上傳文件）

④　查看目錄及文件信息

⑤　查找某個文件在HDFS集羣的位置

⑥　刪除數據

⑦　獲取HDFS集羣上全部數據節點信息

l  在HDFS上建立目錄

 

l  經過FileSystem API讀取數據（下載文件）

l  寫入數據（上傳文件）

 

l  查看目錄及文件信息

 

l  查找某個文件在HDFS集羣的位置

l  刪除數據

l  獲取HDFS集羣上全部數據節點信息

（三）HDFS的Web Console

 

（四）HDFS的回收站

l  默認回收站是關閉的，能夠經過在core-site.xml中添加fs.trash.interval來打開幵配置時間閥值，例如：

 

l  刪除文件時，實際上是放入回收站/trash

l  回收站裏的文件能夠快速恢復

l  能夠設置一個時間閾值，當回收站裏文件的存放時間超過返個閾值，就被完全刪除，而且釋放佔用的數據塊

l  查看回收站

       hdfs dfs -lsr /user/root/.Trash/Current

l  從回收站中恢復

       hdfs dfs -cp /user/root/.Trash/Current/input/data.txt/input

（五）HDFS的快照

l  一個snapshot(快照)是一個所有文件系統、或者某個目錄在某一時刻的鏡像

l  快照應用在如下場景中：

n  防止用戶的錯誤操做

n  備份

n  試驗/測試

n  災難恢復

l  HDFS的快照操做

n  開啓快照

        hdfs dfsadmin -allowSnapshot /input

n  建立快照

        hdfs dfs -createSnapshot /input backup_input_01

n  查看快照

        hdfs lsSnapshottableDir

n  對比快照

        hdfs snapshotDiff /input backup_input_01 backup_input_02

n  恢復快照

        hdfs dfs -cp /input/.snapshot/backup_input_01/data.txt /input

（六）HDFS的用戶權限管理

l  超級用戶：

n  啓動namenode服務的用戶就是超級用戶, 該用戶的組是supergroup

 

l  Shell命令變動

命令	說明
chmod [-R] mode file …	只有文件的全部者或者超級用戶纔有權限改變文件模式。
chgrp [-R] group file …chgrp [-R] group file …	使用chgrp命令的用戶必須屬於特定的組且是文件的全部者，或者用戶是超級用戶
chown [-R] [owner][:[group]] file …	文件的全部者的只能被超級用戶更改。
ls(r) file …	輸出格式作了調整以顯示全部者、組和模式。

 

l  文件系統API變動

（七）HDFS的配額管理

l  什麼是配額？

n  配額就是HDFS爲每一個目錄分配的大小空間，新創建的目錄是沒有配額的，最大的配額是Long.Max_Value。配額爲1能夠強制目錄保持爲空。

l  配額的類型

n  名稱配額

u  用於設置該目錄中可以存放的最多文件（目錄）個數。

n  空間配額

u  用於設置該目錄中最大可以存放的文件大小。

l  配額的應用案例

n  設置名稱配額

u  命令：dfsadmin -setQuota <N> <directory>...<directory>

          　 dfsadmin -clrQuota <directory>...<directory>

u  示例：設置/input目錄的名稱配額爲3

               hdfs dfsadmin -setQuota 3 /input

               清除/input目錄的名稱配額

               hdfs dfsadmin -clrQuota  /input

n  設置空間配額

u  命令：dfsadmin -setSpaceQuota <quota> <dirname>...<dirname>

               dfsadmin -clrSpaceQuota <dirname>...<dirname>

u  示例：設置/input目錄的空間配額爲1M

               hdfs dfsadmin -setSpaceQuota 1048576 /input

               清除input目錄的空間配額

               hdfs dfsadmin -clrSpaceQuota /input

l  注意：若是hdfs文件系統中的文件個數或者大小超過了配額限制，會出現錯誤。

（八）HDFS的安全模式

l  何時安全模式？

安全模式是hadoop的一種保護機制，用於保證集羣中的數據塊的安全性。若是HDFS處於安全模式，則表示HDFS是隻讀狀態。

l  當集羣啓動的時候，會首先進入安全模式。當系統處於安全模式時會檢查數據塊的完整性。假設咱們設置的副本數（即參數dfs.replication）是5，那麼在datanode上就應該有5個副本存在，假設只存在3個副本，那麼比例就是3/5=0.6。在配置文件hdfs-default.xml中定義了一個最小的副本的副本率0.999，如圖：

       

咱們的副本率0.6明顯小於0.99，所以系統會自動的複製副本到其餘的dataNode,使得副本率不小於0.999.若是系統中有8個副本，超過咱們設定的5個副本，那麼系統也會刪除多餘的3個副本。

l  雖然不能進行修改文件的操做，可是能夠瀏覽目錄結構、查看文件內容的。

l  在命令行下是能夠控制安全模式的進入、退出和查看的。

n  命令 hdfs dfsadmin -safemode get 查看安全模式狀態

n  命令 hdfs dfsadmin -safemode enter  進入安全模式狀態

n  命令 hdfs dfsadmin -safemode leave 離開安全模式

（九）HDFS的底層原理

l  HDFS的底層通訊原理採用的是：RPC和動態代理對象Proxy

l  RPC

n  什麼是RPC？

Remote Procedure Call，遠程過程調用。也就是說，調用過程代碼並非在調用者本地運行，而是要實現調用者與被調用者二地之間的鏈接與通訊。

RPC的基本通訊模型是基於Client/Server進程間相互通訊模型的一種同步通訊形式；它對Client提供了遠程服務的過程抽象，其底層消息傳遞操做對Client是透明的。

在RPC中，Client便是請求服務的調用者(Caller)，而Server則是執行Client的請求而被調用的程序 (Callee)。

n  RPC示例

l  服務器端

 

 

l  客戶端

 

l  Java動態代理對象

l  爲其餘對象提供一種代理以控制對這個對象的訪問。

l  核心是使用JDK的Proxy類

7、MapReduce

（一）MapReduce在Yarn平臺上 運行過程

 

（二）第一個MapReduce程序：WordCount

l  Mapper的實現

l  Reducer的實現 

 

l  主程序的實現

 

（三）WordCount的數據流過程

（四）使用MapReduce處理數據

　　emp.csv

①　排序：注意排序按照Key2排序

          須要實現WritableComparable接口

 　　　　　　SortMapper.java　　Employee.java　　SortMain.java

②　分區

 

EmployeePartition.java　　EmployeeReduce.java　　PartitionMain.java　　Employee.java　　EmployeeMapper.java

③　合併：Combiner是一個特殊的Reducer。但不是全部的狀況均可以使用Combiner。

（五）Shuffle的過程

（六）使用MRUnit進行單元測試過程

l  使用的時候須要從官網http://mrunit.apache.org/下載jar包

l  基本原理是JUnit和EasyMock，其核心的單元測試依賴於JUnit，而且MRUnit實現了一套Mock對象來控制MapReduce框架的輸入和輸出；語法也比較簡單。

l  注意：須要把mockito-all-1.8.5.jar從Build Path中去掉

l  以WordCount僞列：

n  測試Mapper

 

n  測試Reducer

n  測試MapReducer

（七）MapReduce做業任務的管理

l  經過Web Console監控做業的運行

l  經過yarn application命令來進行做業管理

①　列出幫助信息：

yarn application --help

②　查看運行Mapreduce程序：

yarn application -list

③　查看應用狀態：

yarn application -status <application_id>

④　強制殺死應用：

yarn application –kill <application_id>

（八）MapReduce案例集錦

l  經過下面六個案例，強化MapReduce程序的開發和理解

n  數據去重

n  數據排序

n  平均成績

n  單表關聯

 

n  多表關聯

n  倒排索引

數據：

結果：

l  數據去重

「數據去重」主要是爲了掌握和利用 並行化思想來對數據進行 有意義的 篩選。統計大數據集上的數據種類個數統計大數據集上的數據種類個數、 從網站日誌中計算訪問地等這些看似龐雜的任務都會涉及數據去重。

一、問題描述

對數據文件中的數據進行去重。數據文件中的每行都是一個數據。

樣例輸入文件以下所示：

文件1： 2012-3-1 a 2012-3-2 b 2012-3-3 c 2012-3-4 d 2012-3-5 a 2012-3-6 b 2012-3-7 c 2012-3-3 c

文件2： 2012-3-1 b 2012-3-2 a 2012-3-3 b 2012-3-4 d 2012-3-5 a 2012-3-6 c 2012-3-7 d 2012-3-3 c

 

樣例輸出以下所示：

結果： 2012-3-1 a 2012-3-1 b 2012-3-2 a 2012-3-2 b 2012-3-3 b 2012-3-3 c 2012-3-4 d 2012-3-5 a 2012-3-6 b 2012-3-6 c 2012-3-7 c 2012-3-7 d

 

二、設計思路

數據去重的最終目標是讓 原始數據中 出現次數 超過一次的 數據在 輸出文件中 只出現一次出現一次。咱們天然而然會想到將同一個數據的全部記錄都交給臺 一臺 reduce 機器，不管這個數據出現多少次，只要在最終結果中輸出一次就能夠了。具體就是 reduce 的 輸入應該以 數據做爲 key，而對 value-list 則 沒有要求。當 reduce 接收到一個<key，value-list>時就 直接將 key複製到輸出的 key 中，並將 value 設置成 空值。

在 MapReduce 流程中， map 的輸出<key， value>通過 shuffle 過程彙集成<key， value-list>後會交給 reduce。因此從設計好的 reduce 輸入能夠反推出 map 的輸出 key 應爲數據，value任意。繼續反推，map 輸出數據的 key 爲數據，而在這個實例中每一個數據表明輸入文件中的一行內容，因此 map 階段要完成的任務就是在採用 Hadoop 默認的做業輸入方式以後，將value 設置爲 key，並直接輸出（輸出中的 value 任意） 。map 中的結果通過 shuffle 過程以後交給 reduce。reduce 階段不會管每一個 key 有多少個 value，它直接將輸入的 key 複製爲輸出的 key，並輸出就能夠了（輸出中的 value 被設置成空了） 。

三、程序代碼

略：

l  數據排序

「 數據排序」是許多實際任務執行時要完成的第一項工做，好比 學生成績評比、數據創建索引數據創建索引等。這個實例和數據去重相似，都是先對原始數據進行 初步處理，爲 進一步的數據操做打好基礎。

一、問題描述

對輸入文件中數據進行排序。輸入文件中的每行內容均爲一個數字，即一個數據。要求在輸出中每行有 兩個間隔的數字，其中， 第一個表明原始數據在原始數據集中的位次，第二個表明原始數據。

輸入文件：

文件1： 2 32 654 32 15 756 65223

文件2： 5956 22 650 92

文件3： 26 54 6

 

輸出文件：

輸出結果： 1  2 2  6 3  15 4  22 5  26 6  32 7  32 8  54 9  92 10 650 11 654 12 756 13 5956 14 6522

 

二、設計思路

該示例僅僅要求對輸入數據進行排序，熟悉 MapReduce 過程的讀者會很快想到在MapReduce過程當中就有排序，是否能夠利用這個 默認的排序，而不須要本身再實現具體的排序呢？答案是確定的。

可是在使用以前首先須要瞭解它的 默認排序規則。它是按照 key 值進行排序的，若是key 爲封裝 int 的 IntWritable 類型，那麼 MapReduce 按照 數字大小對 key 排序，若是 key爲封裝爲 String 的 Text 類型，那麼 MapReduce 按照 字典順序對字符串排序。

瞭解了這個細節，咱們就知道應該使用封裝 int 的 IntWritable 型數據結構了。也就是在map 中將讀入的數據轉化成 IntWritable 型，而後做爲 key 值輸出（value 任意） 。reduce 拿到<key，value-list>以後，將輸入的 key 做爲 value 輸出，並根據 value-list 中 元素的 個數決定輸出的次數。輸出的 key（即代碼中的 linenum）是一個全局變量，它統計當前 key 的位次。須要注意的是這個程序中 沒有配置Combiner， 也就是在MapReduce過程當中不使用Combiner。這主要是由於使用 map 和 reduce 就已經可以完成任務了。

l  平均成績

「平均成績」主要目的仍是在重溫經典「WordCount」例子，能夠說是在基礎上的微變化微變化版，該實例主要就是實現一個計算學平生均成績的例子。

　　1. 問題描述

對輸入文件中數據進行就算學平生均成績。 輸入文件中的每行內容均爲 一個學生的姓名和他相應的 成績，若是有多門學科，則每門學科爲一個文件。要求在輸出中每行有兩個間隔的數據，其中，第一個表明學生的姓名，第二個表明其平均成績。

輸入文件：

數學成績: Tom   88 Mary  99 Mike  66 Jone   77

語文成績: Tom  78 Mary  89 Mike  96 Jone  67

英語成績: Tom   80 Mary   82 Mike   84 Jone   86

 

輸出文件：

平均成績： Tom    82 Mary   90 Mike   82 Jone   76

 

　　2. 設計思路

計算學平生均成績是一個仿「WordCount」例子，用來重溫一下開發 MapReduce 程序的流程。 程序包括兩部分的內容： Map 部分和 Reduce 部分， 分別實現了 map 和 reduce 的功能。

Map處理的是一個純文本文件，文件中存放的數據時每一行表示一個學生的姓名和他相應一科成績。Mapper 處理的數據是由 InputFormat 分解過的數據集，其中 InputFormat 的做用是將數據集切割成小數據集 InputSplit，每個 InputSlit 將由一個 Mapper 負責處理。此外， InputFormat中還提供了一個RecordReader的實現， 並將一個InputSplit解析成<key,value>對提供給了 map 函數。InputFormat 的默認值是 TextInputFormat，它針對文本文件，按行將文本切割成 InputSlit，並用 LineRecordReader 將 InputSplit 解析成<key,value>對，key 是行在文本中的位置，value 是文件中的一行。

Map 的結果會經過 partion 分發到 Reducer，Reducer 作完 Reduce 操做後，將經過以格式 OutputFormat 輸出。

Mapper 最終處理的結果對<key,value>，會送到 Reducer 中進行合併，合併的時候，有相同 key 的鍵/值對則送到同一個 Reducer 上。Reducer 是全部用戶定製 Reducer 類地基礎，它的輸入是 key 和這個 key 對應的全部 value 的一個迭代器，同時還有 Reducer 的上下文。Reduce 的結果由 Reducer.Context 的 write 方法輸出到文件中。

l  單表關聯

前面的實例都是在數據上進行一些簡單的處理，爲進一步的操做打基礎。 「 單表關聯」這個實例 要求從 給出的數據中 尋找所 關心的數據，它是對 原始數據所包含信息的 挖掘。下面進入這個實例。

　　1. 問題描述

實例中給出 child-parent（孩子——父母）表，要求輸出 grandchild-grandparent（孫子——爺奶）表。

樣例輸入以下所示：                　　　　　　　　樣例輸出以下所示：

child	parent	grandchild	grandparent
Tom      Tom     Jone       Jone      Lucy       Lucy       Jack       Jack       Terry      Terry      Philip      Philip	Lucy Jack Lucy Jack Mary Ben Alice Jesse Alice Jesse Terry Alma	Tom Tom Jone Jone Tom Tom Jone Jone Philip Philip Mark Mark	Alice Jesse Alice Jesse Mary Ben Mary Ben Alice Jesse Alice Jesse

 

 

 

 

 

 

 

 

 

 

 

 

家族樹狀關係譜：

　　2.設計思路

分析這個實例，顯然須要進行單錶鏈接，鏈接的是 左表的 parent 列和 右表的 child 列，且 左表和 右表是 同一個表。

鏈接結果中 除去鏈接的兩列就是所須要的結果——「grandchild--grandparent」表。要用MapReduce 解決這個實例， 首先應該考慮如何實現 表的 自鏈接； 其次就是 鏈接列的 設置；最後是 結果的 整理。

考慮到 MapReduce 的 shuffle 過程會將相同的 key 會鏈接在一塊兒， 因此能夠將 map 結果的 key 設置成 待鏈接的 列， 而後列中相同的值就天然會鏈接在一塊兒了。 再與最開始的分析聯繫起來：

要鏈接的是左表的 parent 列和右表的 child 列，且左表和右表是同一個表，因此在 map階段將 讀入數據 分割成 child 和 parent 以後，會將 parent 設置成 key，child 設置成 value進行輸出，並做爲 左表；再將 同一對 child 和 parent 中的 child 設置成 key，parent 設置成value 進行輸出，做爲 右表。爲了 區分輸出中的 左右表，須要在輸出的 value 中 再加上左右表左右表的 信息，好比在 value 的 String 最開始處加上 字符 1 表示 左表，加上 字符 2 表示 右表。這樣在 map 的結果中就造成了左表和右表，而後在 shuffle 過程當中完成鏈接。reduce 接收到鏈接的結果，其中每一個 key 的 value-list 就包含了「grandchild--grandparent」關係。取出每一個key 的 value-list 進行解析， 將 左表中的 child 放入一個 數組，  右表中的 parent 放入一個 數組，而後對 兩個數組求笛卡爾積就是最後的結果了。

l  多表關聯

多表關聯和單表關聯相似，它也是經過對原始數據進行必定的處理， 從其中挖掘出關心的信息。

　　1.問題描述

輸入是兩個文件，一個表明 工廠表，包含 工廠名列和 地址編號列；另外一個表明 地址表包 地址名列和 地址編號列。要求從 輸入數據中找出 工廠名和 地址名的 對應關係，輸出「工廠名——地址名」表。

輸入文件：　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　

工廠信息：	地址信息：

 

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　

輸出文件：

　　2.設計思路

多表關聯和單表關聯類似，都相似於 數據庫中的 天然鏈接。相比單表關聯，多表關聯的左右表和鏈接列更加清楚。 因此能夠採用和 單表關聯的 相同的 處理方式， map 識別出輸入的行屬於哪一個表以後，對其進行分割，將 鏈接的列值保存在 key 中， 另外一列和左右表 標識保存在 value 中，而後輸出。reduce 拿到鏈接結果以後，解析 value 內容，根據標誌將左右表內容分開存放，而後求 笛卡爾積，最後直接輸出。

這個實例的具體分析參考單表關聯實例。

l  倒排索引

「 倒排索引」是 文檔檢索系統中 最經常使用的 數據結構，被普遍地應用於 全文搜索引擎。它 主要是用來 存儲某個 單詞（或詞組） 在一個 文檔或一組文檔中的 存儲位置的 映射，即提供了一種 根據內容來查找文檔的 方式。 因爲不是根據文檔來肯定文檔所包含的內容， 而是進行相反的操做，於是稱爲倒排索引（Inverted Index） 。

　　1.問題描述

一般狀況下，倒排索引由一個單詞（或詞組）以及相關的文檔列表組成，文檔列表中的文檔或者是標識文檔的 ID 號，或者是指文檔所在位置的 URL，以下圖所示。

從上圖能夠看出，單詞 1 出如今{文檔 1，文檔 4，文檔 13，……}中，單詞 2 出如今{文檔 3，文檔 5，文檔 15，……}中，而單詞 3 出如今{文檔 1，文檔 8，文檔 20，……}中。在 實際應用中， 還須要給 每一個文檔添加一個 權值，用來 指出每一個文檔與搜索內容的相關度，以下圖所示：

最經常使用的是使用詞頻做爲權重，即記錄單詞在文檔中出現的次數。以英文爲例，以下圖 所示，索引文件中的「MapReduce」一行表示： 「MapReduce」這個單詞在文本 T0 中出現過 1 次， T1 中出現過 1 次， T2 中出現過 2 次。 當搜索條件爲 「MapReduce」 、 「is」 、 「Simple」時，對應的集合爲：{T0，T1，T2}∩{T0，T1}∩{T0，T1}={T0，T1}，即文檔 T0 和 T1 包含了所要索引的單詞，並且只有 T0 是連續的。

 

更復雜的權重還可能要記錄單詞在多少個文檔中出現過，以實現 TF-IDF（TermFrequency-Inverse Document Frequency）算法，或者考慮單詞在文檔中的位置信息（單詞是否出如今標題中，反映了單詞在文檔中的重要性）等。

輸入文件以下：

 

文件1： MapReduce is simple

文件2： MapReduce is powerful is simple

文件3： Hello MapReduce bye MapReduce

 

輸出文件以下：

　　2.設計思路

實現「 倒排索引」只要關注的信息爲： 單詞、 文檔 URL 及 詞頻。可是在實現過程當中，索引文件的格式與圖 6.1-3 會略有所不一樣，以免重寫 OutPutFormat 類。下面根據 MapReduce  的處理過程給出 倒排索引的 設計思路。

l  Mapper的過程

l  Combiner的過程

l  Reducer的過程

通過上述兩個過程後， Reduce 過程只需將相同 key 值的 value 值組合成倒排索引文件所需的格式便可，剩下的事情就能夠直接交給 MapReduce 框架進行處理了。

（十）搭建Hadoop的Eclipse開發環

hadoop-eclipse-plugin-2.4.1.jar

 

l  設置環境變量：HADOOP_HOME，並把下面的bin目錄加入PATH路徑中：

l  將hadoop.dll文件複製到C:\Windows\System32目錄下

l  修改Hadoop的源碼：org.apache.hadoop.io.nativeio.NativeIO

8、Hive

（一）什麼是Hive

l  構建在Hadoop上的數據倉庫平臺，爲數據倉庫管理提供了許多功能

l  起源自facebook由Jeff Hammerbacher領導的團隊

l  2008年facebook把hive項目貢獻給Apache

l  定義了一種類SQL語言HiveQL。能夠當作是仍SQL到Map-Reduce的映射器

l  提供Hive shell、JDBC/ODBC、Thrift客戶端等接

（二）Hive的體系結構

 

l  用戶接口主要有三個：CLI，JDBC/ODBC和 WebUI

n  CLI，即Shell命令行

n  JDBC/ODBC 是 Hive 的Java，與使用傳統數據庫JDBC的方式相似

n  WebGUI是經過瀏覽器訪問 Hive

l  Hive 將元數據存儲在數據庫中(metastore)，目前只支持 mysql、derby。Hive 中的元數據包括表的名字，表的列和分區及其屬性，表的屬性（是否爲外部表等），表的數據所在目錄等

l  解釋器、編譯器、優化器完成 HQL 查詢語句從詞法分析、語法分析、編譯、優化以及查詢計劃（plan）的生成。生成的查詢計劃存儲在 HDFS 中，並在隨後有 MapReduce 調用執行

l  Hive 的數據存儲在 HDFS 中，大部分的查詢由 MapReduce 完成（包含 * 的查詢，好比 select * from table 不會生成 MapRedcue 任務）

（三）Hive的管理

l  Hive的安裝

n  在虛擬機上安裝MySQL：

u  yum remove mysql-libs

u  rpm -ivh mysql-community-common-5.7.19-1.el7.x86_64.rpm

u  rpm -ivh mysql-community-libs-5.7.19-1.el7.x86_64.rpm

u  rpm -ivh mysql-community-client-5.7.19-1.el7.x86_64.rpm

u  rpm -ivh mysql-community-server-5.7.19-1.el7.x86_64.rpm

u  rpm -ivh mysql-community-devel-5.7.19-1.el7.x86_64.rpm  （可選）

　　啓動MySQL：service mysqld start

       或者：systemctl start mysqld.service

       查看root用戶的密碼：cat /var/log/mysqld.log | grep password

       登陸後修改密碼：alter user 'root'@'localhost' identified by 'Welcome_1';

n  MySQL數據庫的配置：

u  建立一個新的數據庫：create database hive;

u  建立一個新的用戶：create user 'hiveowner'@'%' identified by ‘Welcome_1’;

u  給該用戶受權

        grant all on hive.* TO 'hiveowner'@'%';

        grant all on hive.* TO 'hiveowner'@'localhost' identified by 'Welcome_1';

n  嵌入式模式（本地模式）

u  元數據信息存儲在內置的Derby數據庫中

u  只運行創建一個鏈接

u  設置Hive的環境變量

u  設置如下參數：

參數文件	配置參數	參考值
hive-site.xml	javax.jdo.option.ConnectionURL	jdbc:derby:;databaseName=metastore_db;create=true
	javax.jdo.option.ConnectionDriverName	org.apache.derby.jdbc.EmbeddedDriver
	hive.metastore.local	true
	hive.metastore.warehouse.dir	file:///root/training/apache-hive-2.3.0-bin/warehouse

 

u  初始化MetaStore：

schematool -dbType derby -initSchema

n  遠程模式

u  元數據信息存儲在遠程的MySQL數據庫中

u  注意必定要使用高版本的MySQL驅動（5.1.43以上的版本）

　　mysql-connector-java-5.1.43-bin.jar

參數文件	配置參數	參考值
hive-site.xml	javax.jdo.option.ConnectionURL	jdbc:mysql://localhost:3306/hive?useSSL=false
	javax.jdo.option.ConnectionDriverName	com.mysql.jdbc.Driver
	javax.jdo.option.ConnectionUserName	hiveowner
	javax.jdo.option.ConnectionPassword	Welcome_1

　　u  初始化MetaStore：

　　　　schematool -dbType mysql -initSchema

l  Hive的管理

n  CLI（Command Line Interface）方式

n  HWI（Hive Web Interface）方式（只在2.2.0之前的版本纔有）

u  安裝步驟：

l  tar -zxvf apache-hive-0.13.0-src.tar.gz

l  jar cvfM0 hive-hwi-0.13.0.war -C web/ .

l  修改hive-site.xml

參數文件	配置參數	參考值
hive-site.xml	hive.hwi.listen.host	0.0.0.0
	hive.hwi.listen.port	9999
	hive.hwi.war.file	lib/hive-hwi-0.13.0.war

 

l  注意：拷貝JDK/lib/tools.jar ----> $HIVE_HOME/lib

u  啓動方式：hive --service hwi &

u  用於經過瀏覽器來訪問hive：http://<host ip>:9999/hwi/

n  遠程服務：端口號10000

u  啓動方式：hive --service hiveserver &

（四）Hive的數據類型

l  基本數據類型

n  tinyint/smallint/int/bigint：整數類型

n  float/double：浮點數類型

n  boolean：布爾類型

n  string：字符串類型

l  複雜數據類型

n  Array：數組類型，由一系列相同數據類型的元素組成

n  Map：集合類型，包含key->value鍵值對，能夠經過key來訪問元素。

n  Struct：結構類型，能夠包含不一樣數據類型的元素。這些元素能夠經過」點語法」的方式來獲得所須要的元素

l  時間類型

n  Date：從Hive0.12.0開始支持

n  Timestamp：從Hive0.8.0開始支持

（五）Hive的數據模型

l  Hive的數據存儲

n  基於HDFS

n  沒有專門的數據存儲格式

n  存儲結構主要包括：數據庫、文件、表、視圖

n  能夠直接加載文本文件（.txt文件）

n  建立表時，指定Hive數據的列分隔符與行分隔符

l  表

n  Inner Table（內部表）

u  與數據庫中的 Table 在概念上是相似

u  每個 Table 在 Hive 中都有一個相應的目錄存儲數據

u  全部的 Table 數據（不包括 External Table）都保存在這個目錄中

u  刪除表時，元數據與數據都會被刪除

 

n  Partition Table （分區表）

u  Partition 對應於數據庫的 Partition 列的密集索引

u  在 Hive 中，表中的一個 Partition 對應於表下的一個目錄，全部的 Partition 的數據都存儲在對應的目錄中

 

u  往分區表中插入數據：

u  在Hive中，經過SQL的執行計劃獲知分區表提升的效率

n  External Table（外部表）

u  指向已經在 HDFS 中存在的數據，能夠建立 Partition

u  它和內部表在元數據的組織上是相同的，而實際數據的存儲則有較大的差別

u  外部表 只有一個過程，加載數據和建立表同時完成，並不會移動到數據倉庫目錄中，只是與外部數據創建一個連接。當刪除一個外部表時，僅刪除該連接

 

n  Bucket Table （桶表）

u  桶表是對數據進行哈希取值，而後放到不一樣文件中存儲。

u  須要設置環境變量：set hive.enforce.bucketing = true;

 

l  視圖（View）

n  視圖是一種虛表，是一個邏輯概念；能夠跨越多張表

n  視圖創建在已有表的基礎上, 視圖賴以創建的這些表稱爲基表

n  視圖能夠簡化複雜的查詢

 

（六）Hive數據的導入

l  Hive支持兩種方式的數據導入

l  使用load語句導入數據

l  使用sqoop導入關係型數據庫中的數據

l  使用load語句導入數據

l  數據文件： 

l  導入本地的數據文件

 

注意：Hive默認分隔符是: tab鍵。因此須要在建表的時候，指定分隔符。

 

l  導入HDFS上的數據

 

l  使用sqoop導入關係型數據庫中的數據

n  將關係型數據的表結構複製到hive中

sqoop create-hive-table --connect jdbc:mysql://localhost:3306/test --table username --username root --password 123456 --hive-table test

其中 --table username爲mysql中的數據庫test中的表   --hive-table test 爲hive中新建的表名稱

n  從關係數據庫導入文件到hive中

sqoop import --connect jdbc:mysql://localhost:3306/test --username root --password mysql-password --table t1 --hive-import

n  將hive中的表數據導入到mysql中

sqoop export --connect jdbc:mysql://localhost:3306/test --username root --password admin --table uv_info --export-dir /user/hive/warehouse/uv/dt=2011-08-03

（七）Hive的查詢

l  簡單查詢

l  過濾和排序

l  Hive的函數

　　n  數學函數

　　　　u  round

　　　　u  ceil

　　　　u  floor

　　n  字符函數

　　　　u  lower

　　　　u  upper

　　　　u  length

　　　　u  concat

　　　　u  substr

　　　　u  trim

　　　　u  lpad

　　　　u  rpad

　　n  收集函數

　　　　u  size

　　n  日期函數

　　　　u  to_date

　　　　u  year

　　　　u  month

　　　　u  day

　　　　u  weekofyear

　　　　u  datediff

　　　　u  date_add

　　　　u  date_sub

　　n  條件函數

　　　　u  if

　　　　u  coalesce

　　　　u  case... when...

　　n  聚合函數

　　　　u  count

　　　　u  sum

　　　　u  min

　　　　u  max

　　　　u  avg

l  Hive的多表查詢

n  只支持：等鏈接，外鏈接，左半鏈接

n  不支持非相等的join條件

n  https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Joins

l  Hive的子查詢

n  hive只支持：from和where子句中的子查詢

n  https://cwiki.apache.org/confluence/display/Hive/LanguageManual+SubQueries

（八）Hive的客戶端操做：JDBC

l  首先啓動Hive遠程服務：hiveserver2 &

l  須要Hive lib目錄下的jar包

l  演示案例：查詢數據

 TestMain.java　　JDBCUtils.java

l  若是出現如下錯誤：

修改hadoop 配置文件 etc/hadoop/core-site.xml,加入以下配置項

（十）Hive的自定義函數

l  Hive的自定義函數（UDF）： User Defined Function

n  能夠直接應用於select語句，對查詢結構作格式化處理後，再輸出內容

l  Hive自定義函數的實現細節

n  自定義UDF須要繼承org.apache.hadoop.hive.ql.UDF

n  須要實現evaluate函數，evaluate函數支持重載

l  Hive自定義函數案例

n  案例一：拼加兩個字符串

n  案例二：判斷員工表中工資的級別

l  Hive自定義函數的部署

n  把程序打包放到目標機器上去

n  進入hive客戶端，添加jar包：

hive> add jar /root/temp/udf.jar;

n  建立臨時函數：

l  Hive自定義函數的調用

n  查詢HQL語句：

select myconcat(ename,job) from emp;

select ename,sal,checksalary(sal) from emp;

n  銷燬臨時函數：

hive> DROP TEMPORARY FUNCTION checksalary;

9、Pig

（一）什麼是Pig？

l  Pig是一個用來處理大規模數據集的平臺，由Yahoo！貢獻給Apache

l  Pig能夠簡化MapReduce任務的開發

l  Pig能夠看作hadoop的客戶端軟件，能夠鏈接到hadoop集羣迕行數據分析工做

l  Pig方便不熟悉java的用戶，使用一種較爲簡便的相似二SQL的面向數據流的語言pig Latin迕行數據處理

l  Pig Latin能夠迕行排序、過濾、求和、分組、關聯等經常使用操做，迓能夠自定義凼數，返是一種面向數據分析處理的輕量級腳本語言

l  Pig能夠看作是Pig Latin到MapReduce的映射器

l  Pig能夠自動對集羣迕行分配和回收，幵丏自勱地對MapReduce程序迕行優化

（二）Pig的體系結構

 

（三）Pig的安裝和工做模式

l  安裝步驟

n  下載並解壓pig安裝包（ http://pig.apache.org/）

n  設置環境變量

l  工做模式

n  本地模式：pig -x local

n  MapReduce模式

u  設置PIG_CLASSPATH環境變量，指向Hadoop的的配置目錄

u  啓動：pig

l  Pig的經常使用命令

n  ls、cd、cat、mkdir、pwd

n  copyFromLocal、copyToLocal

n  sh

n  register、define

（四）Pig的內置函數

l  計算函數

avg	計算包中項的平均值
concat	把兩個字節數組或者字符數組鏈接成一個
count	計算包中非空值的個數
count_star	計算包中項的個數，包括空值
diff	計算兩個包的差
max	計算包中項的最大值
min	計算包中項的最小值
size	計算一個類型的大小，數值型的大小爲1； 對於字符數組，返回字符的個數； 對於字節數組，返回字節的個數； 對於元組，包，映射，返回其中項的個數。
sum	計算一個包中項的值的總和
TOKENIZE	對一個字符數組進行標記解析，並把結果詞放入一個包

l  過濾函數

isempty

判斷一個包或映射是否爲空

l  加載存儲函數

PigStorage	用字段分隔文本格式加載或存儲關係，這是默認的存儲函數
BinStorage	從二進制文件加載一個關係或者把關係存儲到二進制文件
BinaryStorage	從二進制文件加載只是包含一個類型爲bytearray的字段的元組到關係，或以這種格式存儲一個關係
TextLoader	從純文本格式加載一個關係
PigDump	用元組的tostring()形式存儲關係

（五）使用Pig Latin語句分析數據

l  Pig的數據模型

Bag：表

Tuple：行，記錄

Field：屬性

Pig不要求同一個bag裏面的各個tuple有相同數量或相同類型的field

l  注意：啓動historyserver：

mr-jobhistory-daemon.sh start historyserver

能夠經過：http://<IP>:19888/jobhistory查看歷史任務記錄

l  經常使用的Pig Latin語句

n  LOAD：指出載入數據的方法

 

n  FOREACH：逐行掃描進行某種處理

對數據表進行逐行處理是十分重要的功能，所以PIG Latin提供了FOREACH命令。經過一組表達式對數據流中的每一行進行運算，產生的結果就是用於下一個算子的數據集。

下面的語句加載整個數據，但最終結果B中只保留其中的user和id字段：

表達式中可使用「*」表明所有列，還可使用「..」表示範圍內的列，這對簡化命令文本頗有用：

n  FILTER：過濾行

使用FILTER能夠對數據表中的數據進行過濾工做。該關鍵字常常與 BY 一塊兒聯用，對錶中對應的特徵值進行過濾，例如：

n  GROUP分組

按照鍵值相同的規則歸併數據。在SQL中，group操做必然與彙集函數組合使用，而在pig中，group操做將產生與鍵值有關的bag。

這裏的group是默認的變量，表示使用GROUP聚合後獲得的數據組，再經過 FOREACH 對每一組數據進行單獨處理。

n  Distinct 剔除重複項

數據表中若存在重複項，而且須要剔除這些數據，那麼可使用DISTINCT命令，剔除數據表中重複項，該方法與SQL中的用法一致。

n  Union 聯合

Union 的做用與 SQL 中的 Union 類似，使用該操做符，能夠將不一樣的數據表，組合成同一的數據表。假設咱們有數據表A和表B，都記錄了行車數據：

n  Join 連接

 

n  DUMP：把結果顯示到屏幕

n  STORE：把結果保存到文件

l  WordCount實現

① 加載數據

mydata = load '/data/data.txt' as (line:chararray);

② 將字符串分割成單詞

words = foreach mydata generate flatten(TOKENIZE(line)) as word;

③ 對單詞進行分組

grpd = group words by word;

④ 統計每組中單詞數量

cntd = foreach grpd generate group,COUNT(words);

⑤ 打印結果

dump cntd;

（六）Pig的自定義函數

l  概述

n  支持使用Java、Python、Javascript三種語言編寫UDF

n  Java自定義函數較爲成熟，其它兩種功能還有限

n  須要的jar包：

u  /root/training/pig-0.14.0/pig-0.14.0-core-h2.jar

u  /root/training/pig-0.14.0/lib

u  /root/training/pig-0.14.0/lib/h2

u  /root/training/hadoop-2.4.1/share/hadoop/common

u  /root/training/hadoop-2.4.1/share/hadoop/common/lib

l  Pig函數的類型

n  自定義過濾函數

n  自定義運算函數

n  自定義加載函數

l  部署自定義函數

n  註冊jar包  register /root/temp/pig.jar

n  爲自定義函數起別名：define myfilter demo.pig.MyFilterFunction;

l  自定義過濾函數

n  示例：若是員工工資大於等於3000塊錢，則被選擇出來。

 

n  用法：filter emp by demo.pig.IsSalaryTooHigh(sal)

n  也能夠建立函數的別名

l  自定義運算函數

n  判斷員工薪水是級別

u  若是sal <= 1000，則爲：Grade A

u  若是sal > 1000 && sal<=3000，則爲：Grade B

u  若是sal > 3000，則爲：Grade C

n  參考運行結果

 

n  用法：emp1 = foreach emp generate ename,demo.pig.CheckSalaryGrade(sal);

l  自定義加載函數

n  默認狀況下，一行數據會被解析成一個Tuple

n  好比：員工信息

 

n  特殊狀況：單詞統計的時候。這時候：但願每一個單詞能被解析成一個Tuple，從而便於處理

 

n  須要MapReduce的庫（jar文件）

 

n  運行加載函數：records = load '/input/data.txt' using demo.pig.MyLoadFunction();

10、HBase

（一）什麼是HBase？

HBase是一個分佈式的、面向列的開源數據庫，該技術來源於 Fay Chang 所撰寫的Google論文「Bigtable：一個結構化數據的分佈式存儲系統」。就像Bigtable利用了Google文件系統（File System）所提供的分佈式數據存儲同樣，HBase在Hadoop之上提供了相似於Bigtable的能力。HBase是Apache的Hadoop項目的子項目。HBase不一樣於通常的關係數據庫，它是一個適合於非結構化數據存儲的數據庫。另外一個不一樣的是HBase基於列的而不是基於行的模式。

（二）HBase的體系結構

（三）HBase的表結構

（四）HBase的安裝和部署

l  本地模式

l  僞分佈模式

l  集羣模式

本地模式的配置

參數文件	配置參數	參考值
.bash_profile	HBASE_HOME	/root/training/hbase-1.3.1
hbase-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
hbase-site.xml	hbase.rootdir	file:///root/training/hbase-1.3.1/data

 

僞分佈模式

參數文件	配置參數	參考值
.bash_profile	HBASE_HOME	/root/training/hbase-1.3.1
hbase-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
	HBASE_MANAGES_ZK	true
hbase-site.xml	hbase.rootdir	hdfs://192.168.157.111:9000/hbase
	hbase.cluster.distributed	true
	hbase.zookeeper.quorum	192.168.157.111
	dfs.replication	1
regionservers		192.168.157.111

 

集羣模式

參數文件	配置參數	參考值
.bash_profile	HBASE_HOME	/root/training/hbase-1.3.1
hbase-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
	HBASE_MANAGES_ZK	true
hbase-site.xml	hbase.rootdir	hdfs://192.168.157.111:9000/hbase
	hbase.cluster.distributed	true
	hbase.zookeeper.quorum	192.168.157.111
	dfs.replication	2
	hbase.master.maxclockskew	180000
regionservers		192.168.157.113 192.168.157.114

 

注意每臺機器的時間，可使用 date -s 11/22/2016 設置時間

HBase Web Console（端口：16010）

（五）-ROOT-和.META.

n  HBase中有兩張特殊的Table，-ROOT-和.META.

u  -ROOT- ：記錄了.META.表的Region信息，-ROOT-只有一個region

u  .META. ：記錄了用戶建立的表的Region信息，.META.能夠有多個regoin

n  Zookeeper中記錄了-ROOT-表的location

n  Client訪問用戶數據以前須要首先訪問zookeeper，而後訪問-ROOT-表，接着訪問.META.表，最後才能找到用戶數據的位置去訪問。

（六）HBase Shell

n  HBase提供了一個shell的終端給用戶交互

（七）HBase的Java編程接口

l  建立表 

l  刪除表 

l  插入單條數據 

l  插入多條數據 

l  查詢數據 

l  掃描表的數據 

（八）HBase上的過濾器

　　　　　　　　　　DataInit.java

l  列值過濾器

l  列名前綴過濾器 

l  多個列名前綴過濾器 

l  Rowkey過濾器 

l  在查詢中使用多個過濾器 

（九）HBase上的MapReduce

l  實現WordCount

創建測試數據：

create 'word','content'

put 'word','1','content:info','I love Beijing'

put 'word','2','content:info','I love China'

put 'word','3','content:info','Beijing is the capital of China'

create 'stat','content'

l  Mapper

l  Reducer

l  主程序

l  export HADOOP_CLASSPATH=$HBASE_HOME/lib/*:$CLASSPATH

l  練習：求每一個部門的工資總額。

（十）HBase的HA

l  架構

 

l  配置參數：hbase.zookeeper.quorum

l  單獨啓動HMaster：hbase-daemon.sh start master 

11、Sqoop

（一）什麼是Sqoop？

Sqoop是一款開源的工具，主要用於在Hadoop(Hive)與傳統的數據庫(mysql、postgresql...)間進行數據的傳遞，能夠將一個關係型數據庫（例如 ： MySQL ,Oracle ,Postgres等）中的數據導進到Hadoop的HDFS中，也能夠將HDFS的數據導進到關係型數據庫中。

Sqoop項目開始於2009年，最先是做爲Hadoop的一個第三方模塊存在，後來爲了讓使用者可以快速部署，也爲了讓開發人員可以更快速的迭代開發，Sqoop獨立成爲一個Apache項目。

（二）Sqoop是如何工做？

u  利用JDBC鏈接關係型數據庫

u  安裝包：sqoop-1.4.5.bin__hadoop-0.23.tar.gz

（三）使用Sqoop

命令	說明
codegen	將關係數據庫表映射爲一個Java文件、Java class類、以及相關的jar包
create-hive-table	生成與關係數據庫表的表結構對應的HIVE表
eval	以快速地使用SQL語句對關係數據庫進行操做，這可使得在使用import這種工具進行數據導入的時候，能夠預先了解相關的SQL語句是否正確，並能將結果顯示在控制檯。
export	從hdfs中導數據到關係數據庫中
help
import	將數據庫表的數據導入到HDFS中
import-all-tables	將數據庫中全部的表的數據導入到HDFS中
job	用來生成一個sqoop的任務，生成後，該任務並不執行，除非使用命令執行該任務。
list-databases	打印出關係數據庫全部的數據庫名
list-tables	打印出關係數據庫某一數據庫的全部表名
merge	將HDFS中不一樣目錄下面的數據合在一塊兒，並存放在指定的目錄中
metastore	記錄sqoop job的元數據信息
version	顯示sqoop版本信息

 

注意：大小寫

實例一：

sqoop codegen --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --table EMP

也能夠寫成下面的形式：

sqoop codegen \

--connect jdbc:oracle:thin:@192.168.137.129:1521:orcl \

--username SCOTT --password tiger \

--table EMP

實例二：

sqoop create-hive-table --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --table EMP --hive-table emphive

實例三：

sqoop eval --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --query "select ename from emp where deptno=10"

sqoop eval --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --query "select ename,dname from emp,dept where emp.deptno=dept.deptno"

實例四：

sqoop export --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --table STUDENTS --export-dir /students

實例五：

sqoop import --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --table EMP --target-dir /emp

實例六：

sqoop import-all-tables --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger -m 1

實例七：

sqoop list-databases --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SYSTEM --password password

實例八：

sqoop list-tables --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger

實例九：

sqoop version

實例十：將數據導入HBase（須要事先將表建立）

sqoop import --connect jdbc:oracle:thin:@192.168.137.129:1521:orcl --username SCOTT --password tiger --table EMP --columns empno,ename,sal,deptno --hbase-table emp --hbase-row-key empno --column-family empinfo

12、Flume

（一）什麼是Flume？

Flume是Cloudera提供的一個高可用的，高可靠的，分佈式的海量日誌採集、聚合和傳輸的系統，Flume支持在日誌系統中定製各種數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各類數據接受方（可定製）的能力。 

　　　　MyDemoWeb.war

（二）Flume的體系結構

 

（三）安裝和配置Flume

         一、解壓

         二、修改conf/flume-env.sh設置JAVA_HOME便可

（四）使用Flume採集日誌數據

案例一：

#bin/flume-ng agent -n a1 -f myagent/a1.conf -c conf -Dflume.root.logger=INFO,console

#定義agent名， source、channel、sink的名稱

a1.sources = r1

a1.channels = c1

a1.sinks = k1

 

#具體定義source

a1.sources.r1.type = netcat

a1.sources.r1.bind = localhost

a1.sources.r1.port = 8888

 

#具體定義channel

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

 

#具體定義sink

a1.sinks.k1.type = logger

 

#組裝source、channel、sink

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

 

案例二：

#bin/flume-ng agent -n a2 -f myagent/a2.conf -c conf -Dflume.root.logger=INFO,console

#定義agent名， source、channel、sink的名稱

a2.sources = r1

a2.channels = c1

a2.sinks = k1

 

#具體定義source

a2.sources.r1.type = exec

a2.sources.r1.command = tail -f /home/hadoop/a.log

 

#具體定義channel

a2.channels.c1.type = memory

a2.channels.c1.capacity = 1000

a2.channels.c1.transactionCapacity = 100

 

#具體定義sink

a2.sinks.k1.type = logger

 

#組裝source、channel、sink

a2.sources.r1.channels = c1

a2.sinks.k1.channel = c1

案例三：

#bin/flume-ng agent -n a3 -f myagent/a3.conf -c conf -Dflume.root.logger=INFO,console

#定義agent名， source、channel、sink的名稱

a3.sources = r1

a3.channels = c1

a3.sinks = k1

 

#具體定義source

a3.sources.r1.type = spooldir

a3.sources.r1.spoolDir = /root/training/logs1

 

#具體定義channel

a3.channels.c1.type = memory

a3.channels.c1.capacity = 1000

a3.channels.c1.transactionCapacity = 100

 

#具體定義sink

a3.sinks.k1.type = logger

 

#組裝source、channel、sink

a3.sources.r1.channels = c1

a3.sinks.k1.channel = c1

案例四：

#bin/flume-ng agent -n a4 -f myagent/a4.conf -c conf -Dflume.root.logger=INFO,console

#定義agent名， source、channel、sink的名稱

a4.sources = r1

a4.channels = c1

a4.sinks = k1

 

#具體定義source

a4.sources.r1.type = spooldir

a4.sources.r1.spoolDir = /root/training/logs

 

#具體定義channel

a4.channels.c1.type = memory

a4.channels.c1.capacity = 10000

a4.channels.c1.transactionCapacity = 100

 

#定義攔截器，爲消息添加時間戳

a4.sources.r1.interceptors = i1

a4.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

 

 

#具體定義sink

a4.sinks.k1.type = hdfs

a4.sinks.k1.hdfs.path = hdfs://192.168.56.111:9000/flume/%Y%m%d

a4.sinks.k1.hdfs.filePrefix = events-

a4.sinks.k1.hdfs.fileType = DataStream

 

#不按照條數生成文件

a4.sinks.k1.hdfs.rollCount = 0

#HDFS上的文件達到128M時生成一個文件

a4.sinks.k1.hdfs.rollSize = 134217728

#HDFS上的文件達到60秒生成一個文件

a4.sinks.k1.hdfs.rollInterval = 60

 

#組裝source、channel、sink

a4.sources.r1.channels = c1

a4.sinks.k1.channel = c1

十3、HUE

（一）什麼是HUE？

Hue是一個開源的Apache Hadoop UI系統，最先是由Cloudera Desktop演化而來，由Cloudera貢獻給開源社區，它是基於Python Web框架Django實現的。經過使用Hue咱們能夠在瀏覽器端的Web控制檯上與Hadoop集羣進行交互來分析處理數據，例如操做HDFS上的數據，運行MapReduce Job等等。

 

（二）HUE所須要的rpm包

gcc g++ libxml2-devel libxslt-devel cyrus-sasl-devel cyrus-sasl-gssapi

python-devel python-setuptools sqlite-devel ant libsasl2-dev libsasl2-modules-gssapi-mit libkrb5-dev

libtidy-0.99-0 mvn openldap-dev libldap2-dev

 

     步驟：

l  掛載光盤（rhel-server-7.4-x86_64-dvd.iso）：mount /dev/cdrom /mnt

l  建立rpm源文件 vi /etc/yum.repos.d/rhel7.repo

[rhel-yum]

name=rhel7

baseurl=file:///mnt

enabled=1

gpgcheck=0 

l  執行下面的語句：

 yum install gcc g++ libxml2-devel libxslt-devel cyrus-sasl-devel cyrus-sasl-gssapi mysql-devel python-devel python-setuptools sqlite-devel ant ibsasl2-dev libsasl2-modules-gssapi-mit libkrb5-dev libtidy-0.99-0 mvn openldap-dev libffi-devel gmp-devel openldap-devel

l  安裝HUE

n  解壓：

u  tar -xvf hue-4.0.1.tgz

n  編譯安裝：注意系統時間

u  PREFIX=/root/training/ make install

n  添加用戶hue

u  adduser hue

u  chown -R hue.hue /root/training/hue/

（三）HUE與Hadoop集成

l  編輯Hadoop的配置文件

l  修改hue.ini（$HUE_HOME/desktop/conf）參數文件（hue.ini）

l  啓動HUE

啓動Hadoop相關組件

啓動Hue: bin/supervisor  （/root/training/hue/build/env）

訪問首頁：http://192.168.157.111:8888/

 

（四）HUE與HBase集成

l  Hue須要讀取Hbase的數據是使用thrift的方式，默認HBase的thrift服務沒有開啓，全部須要手動額外開啓thrift 服務。

thrift service默認使用的是9090端口，使用以下命令查看端口是否被佔用

netstat -nl|grep 9090

l  啓動HBase thrift service

hbase-daemon.sh start thrift

l  配置HUE

hbase_clusters=(Cluster|192.168.157.11:9090)

hbase_conf_dir=/root/training/hbase-1.3.1/conf

（五）HUE與Hive集成

l  配置HUE

 hive_server_host=192.168.157.11

 hive_server_port=10000

 hive_conf_dir=/root/training/apache-hive-2.3.0-bin/conf

l  啓動Hive

         hive --service metastore

         hiveserver2

（六）HUE的文檔

http://cloudera.github.io/hue/docs-4.0.0/manual.html

十4、ZooKeeper

（一）什麼是ZooKeeper?

ZooKeeper是一個分佈式的，開放源碼的分佈式應用程序協調服務，是Google的Chubby一個開源的實現，是Hadoop和Hbase的重要組件。它是一個爲分佈式應用提供一致性服務的軟件，提供的功能包括：配置維護、域名服務、分佈式同步、組服務等。

（二）ZooKeeper的體系結構

 

（三）Zookeeper能幫咱們作什麼？

l  Hadoop2.0,使用Zookeeper的事件處理確保整個集羣只有一個活躍的NameNode,存儲配置信息等.

l  HBase,使用Zookeeper的事件處理確保整個集羣只有一個HMaster,察覺HRegionServer聯機和宕機,存儲訪問控制列表等.

（四）安裝和配置Zookeeper

一、在主節點（hadoop112）上配置ZooKeeper

（*）配置/root/training/zookeeper-3.4.6/conf/zoo.cfg文件

dataDir=/root/training/zookeeper-3.4.6/tmp

　　server.1=hadoop112:2888:3888

　　server.2=hadoop113:2888:3888

　　server.3=hadoop114:2888:3888

（*）在/root/training/zookeeper-3.4.6/tmp目錄下建立一個myid的空文件

　　echo 1 > /root/training/zookeeper-3.4.6/tmp/myid

（*）將配置好的zookeeper拷貝到其餘節點，同時修改各自的myid文件

　　scp -r /root/training/zookeeper-3.4.6/ hadoop113:/root/training

　　scp -r /root/training/zookeeper-3.4.6/ hadoop114:/root/training

二、啓動ZooKeeper和查看ZooKeeper的狀態

zkServer.sh  start

zkServer.sh  status

（五）操做Zookeeper

l  經過ZooKeeper的命令行

n  zkCli.sh

n  建立一個節點（數據）：create /mydata helloworld

l  Demo：ZooKeeper集羣間信息的同步。

l  Demo：ZooKeeper實現分佈式鎖。

pom.xml　　TestDistributedLock.java

十5、Hadoop的集羣和HA

（一）HDFS的聯盟

l  配置HDFS的聯盟

HDFS的Federation（聯盟）：即 集羣中存在多個namenode，每一個數據節點須要向不一樣的Namenode註冊

參數文件	配置參數	參考值
hadoop-env.sh	JAVA_HOME	/root/training/jdk1.8.0_144
core-site.xml	hadoop.tmp.dir	/root/training/hadoop-2.7.3/tmp
hdfs-site.xml	dfs.nameservices	ns1,ns2
	dfs.namenode.rpc-address.ns1	192.168.157.12:9000
	dfs.namenode.http-address.ns1	192.168.157.12:50070
	dfs.namenode.secondaryhttp-address.ns1	192.168.157.12:50090
	dfs.namenode.rpc-address.ns2	192.168.157.13:9000
	dfs.namenode.http-address.ns2	192.168.157.13:50070
	dfs.namenode.secondaryhttp-address.ns2	192.168.157.13:50090
	dfs.replication	2
	dfs.permissions	false
mapred-site.xml	mapreduce.framework.name	yarn
yarn-site.xml	yarn.resourcemanager.hostname	192.168.157.12
	yarn.nodemanager.aux-services	mapreduce_shuffle
slaves	DataNode的地址	192.168.157.15 192.168.157.14

 

在每一個NameNode上格式化：hdfs namenode -format -clusterId xdl1

l  配置viewFS

修改core-site.xml文件，直接加入如下內容：

         <property>

                   <name>fs.viewfs.mounttable.xdl1.homedir</name>

                   <value>/home</value>

         </property>

         <property>

                   <name>fs.viewfs.mounttable.xdl1.link./movie</name>

                   <value>hdfs://192.168.157.12:9000/movie</value>

         </property>

         <property>

                   <name>fs.viewfs.mounttable.xdl1.link./mp3</name>

                   <value>hdfs://192.168.157.13:9000/mp3</value>

         </property>

         <property>

                   <name>fs.default.name</name>

                   <value>viewfs://xdl1</value>

         </property>

3、建立對應的目錄：

hadoop fs -mkdir hdfs://192.168.157.12:9000/movie

hadoop fs -mkdir hdfs://192.168.157.13:9000/mp3

4、就可使用了

hdfs dfs -put a.txt /movie/a.txt

hdfs dfs -cat /movie/a.txt

（二）利用ZooKeeper實現Hadoop的HA

 

　　　　實現Hadoop的HA.txt

注意：

bigdata113上的ResourceManager須要單獨啓動

         命令：yarn-daemon.sh start resourcemanager

十6、Storm

（一）什麼是Storm?

Storm爲分佈式實時計算提供了一組通用原語，可被用於「流處理」之中，實時處理消息並更新數據庫。這是管理隊列及工做者集羣的另外一種方式。 Storm也可被用於「連續計算」（continuous computation），對數據流作連續查詢，在計算時就將結果以流的形式輸出給用戶。它還可被用於「分佈式RPC」，以並行的方式運行昂貴的運算。

Storm能夠方便地在一個計算機集羣中編寫與擴展複雜的實時計算，Storm用於實時處理，就比如 Hadoop 用於批處理。Storm保證每一個消息都會獲得處理，並且它很快——在一個小集羣中，每秒能夠處理數以百萬計的消息。更棒的是你可使用任意編程語言來作開發。

（二）離線計算和流式計算

①　離線計算

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

l  表明技術：Sqoop批量導入數據、HDFS批量存儲數據、MapReduce批量計算、Hive

②　流式計算

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

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

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

③　Storm與Hadoop的區別

Storm用於實時計算	Hadoop用於離線計算
Storm處理的數據保存在內存中，源源不斷	Hadoop處理的數據保存在文件系統中，一批一批
Storm的數據經過網絡傳輸進來	Hadoop的數據保存在磁盤中
Storm與Hadoop的編程模型類似

（三）Storm的體系結構

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

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

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

l  Executor：Storm 0.8以後，Executor爲Worker進程中的具體的物理線程，同一個Spout/Bolt的Task可能會共享一個物理線程，一個Executor中只能運行隸屬於同一個Spout/Bolt的Task。

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

 

（四）Storm的運行機制

n  整個處理流程的組織協調不用用戶去關心，用戶只須要去定義每個步驟中的具體業務處理邏輯

n  具體執行任務的角色是Worker，Worker執行任務時具體的行爲則有咱們定義的業務邏輯決定

（五）Storm的安裝配置

l  解壓：tar -zxvf apache-storm-1.0.3.tar.gz -C ~/training/

l  設置環境變量

 

l  編輯配置文件：$STORM_HOME/conf/storm.yaml

 

n  注意：若是要搭建Storm的HA，只須要在nimbus.seeds中設置多個nimbus便可。

l  把安裝包複製到其餘節點上。

（六）啓動和查看Storm

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

n  storm nimbus &

n  storm logviewer &

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

n  storm ui &

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

n  storm supervisor &

n  storm logviewer &

l  查看storm集羣：訪問nimbus.host:/8080，便可看到storm的ui界面

（七）Storm的經常使用命令

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

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

 

②　殺死任務命令格式：storm kill 【拓撲名稱】 -w 10

（執行kill命令時能夠經過-w [等待秒數]指定拓撲停用之後的等待時間）

storm kill topology-name -w 10

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

storm deactivte topology-name

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

storm activate topology-name

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

storm rebalance topology-name

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

（八）Demo演示：WordCount及流程分析

經過查看Storm UI上每一個組件的events連接，能夠查看Storm的每一個組件（spout、blot）發送的消息。但Storm的event logger的功能默認是禁用的，須要在配置文件中設置：topology.eventlogger.executors: 1，具體說明以下：

l  "topology.eventlogger.executors": 0      默認，禁用

l  "topology.eventlogger.executors": 1      一個topology分配一個Event Logger.

l  "topology.eventlogger.executors": nil    每一個worker.分配一個Event Logger

WordCount的數據流程分析

（九）Storm的編程模型

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

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

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

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

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

l  Stream：表示數據的流向。

l  StreamGroup：數據分組策略

n  Shuffle Grouping ：隨機分組，儘可能均勻分佈到下游Bolt中

n  Fields Grouping ：按字段分組，按數據中field值進行分組；相同field值的Tuple被髮送到相同的Task

n  All grouping ：廣播

n  Global grouping ：全局分組，Tuple被分配到一個Bolt中的一個Task，實現事務性的Topology。

n  None grouping ：不分組

n  Direct grouping ：直接分組 指定分組

（十）Storm編程案例：WordCount

流式計算通常架構圖：

 

l  Flume用來獲取數據。

l  Kafka用來臨時保存數據。

l  Strom用來計算數據。

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

n  建立Spout（WordCountSpout）組件採集數據，做爲整個Topology的數據源

 

n  建立Bolt（WordCountSplitBolt）組件進行分詞操做

n  建立Bolt（WordCountBoltCount）組件進行單詞計數做

 

n  建立主程序Topology（WordCountTopology），並提交到本地運行

n  也能夠將主程序Topology（WordCountTopology）提交到Storm集羣運行

（十一）Storm集羣在ZK上保存的數據結構

 

（十二）Storm集羣任務提交流程

 

（十三）Storm內部通訊機制

 

（十四）集成Storm

（1）與JDBC集成

l  將Storm Bolt處理的結果插入MySQL數據庫中

l  須要依賴的jar包

n  $STORM_HOME\external\sql\storm-sql-core\*.jar

n  $STORM_HOME\external\storm-jdbc\storm-jdbc-1.0.3.jar

n  mysql的驅動

n  commons-lang3-3.1.jar

mysql-connector-java-5.1.7-bin.jar　　commons-lang3-3.1.jar

l  與JDBC集成的代碼實現

n  修改主程序WordCountTopology，增長以下代碼：

 

        增長一個新方法建立JDBCBolt組件

 

n  實現ConnectionProvider接口

 

n  修改WordCountBoltCount組件，將統計後的結果發送給下一個組件寫入MySQL

 

（2）與Redis集成

Redis是一個key-value存儲系統。和Memcached相似，它支持存儲的value類型相對更多，包括string(字符串)、list(鏈表)、set(集合)、zset(sorted set --有序集合)和hash（哈希類型）。與Memcached同樣，爲了保證效率，數據都是緩存在內存中。區別的是redis會週期性的把更新的數據寫入磁盤或者把修改操做寫入追加的記錄文件，而且在此基礎上實現了master-slave(主從)同步。

Redis 是一個高性能的key-value數據庫。Redis的出現，很大程度補償了memcached這類key/value存儲的不足，在部分場合能夠對關係數據庫起到很好的補充做用。它提供了Java，C/C++，C#，PHP，JavaScript，Perl，Object-C，Python，Ruby，Erlang等客戶端，使用很方便。[1]

Redis支持主從同步。數據能夠從主服務器向任意數量的從服務器上同步，從服務器能夠是關聯其餘從服務器的主服務器。

 集成redis的jar包.zip

l  修改代碼：WordCountTopology.java

 

 

（3）與HDFS集成

u  須要的jar包：

l  $STORM_HOME\external\storm-hdfs\storm-hdfs-1.0.3.jar

l  HDFS相關的jar包

u  開發新的bolt組件

（4）與HBase集成

u  須要的jar包：HBase的相關包

u  開發新的bolt組件（WordCountBoltHBase.java）

 

（5）與Apache Kafka集成

l  注意：須要把slf4j-log4j12-1.6.4.jar包去掉，有衝突（有兩個）

（6）與Hive集成

u  因爲集成Storm和Hive依賴的jar較多，而且衝突的jar包不少，強烈建議使用Maven來搭建新的工程。

<dependencies>

         <dependency>

                   <groupId>org.apache.storm</groupId>

                   <artifactId>storm-core</artifactId>

                   <version>1.0.3</version>

                   <scope>provided</scope>

         </dependency>

         <dependency>

                   <groupId>org.apache.storm</groupId>

                   <artifactId>storm-hive</artifactId>

                   <version>1.0.3</version>

                   <type>jar</type>

         </dependency>

</dependencies>

u  須要對Hive作必定的配置（在hive-site.xml文件中）：

<property> 

<name>hive.in.test</name> 

<value>true</value> 

</property>

u  須要使用下面的語句在hive中建立表：

create table wordcount

(word string,total int)

clustered by (word) into 10 buckets

stored as orc TBLPROPERTIES('transactional'='true');

u  啓動metastore服務：hive --service metastore

u  開發新的bolt組件，用於將前一個bolt處理的結果寫入Hive

u  爲了測試的方便，咱們依然採用以前隨機產生字符串的Spout組件產生數據

（7）與JMS集成

JMS即Java消息服務（Java Message Service）應用程序接口，是一個Java平臺中關於面向消息中間件（MOM）的API，用於在兩個應用程序之間，或分佈式系統中發送消息，進行異步通訊。Java消息服務是一個與具體平臺無關的API，絕大多數MOM提供商都對JMS提供支持。

JMS的兩種消息類型：Queue和Topic

基於Weblogic的JMS架構：

 

 

　　　　　　　　　　　　　　　　　　　　　　　　　　WordCountTopology.java

l  須要的weblogic的jar包

wljmsclient.jar　　wlclient.jar

l  permission javax.management.MBeanTrustPermission "register";