大數(shù)據(jù)分析之技術框架整理
大數(shù)據(jù)離線部分
HDFS
1:HDFS的架構部分及工作原理
NameNode:負責管理元素據(jù),將信息保存在內存中
DataNode:保存數(shù)據(jù),以塊的形式保存。啟動后需要定時的向NameNode發(fā)送心跳,報告自身存儲的塊信息
2:HDFS的上傳過程
3:HDFS的下載
4:NameNode的元數(shù)據(jù)安全機制
以記日志的形式將每一個操作寫在磁盤的日志文件中,然后借助Secondary NameNode的checkpoint功能將fsImage和日志進行合并。
重點:記住checkpoint工作過程
5:如果服務器的磁盤壞了,如何挽救數(shù)據(jù)?
配置多個dfs.namenode.name.dir 路徑為本地磁盤路徑和nfs網(wǎng)絡磁盤路徑。
6:hdfs集群中,受到拓展瓶頸的是NameNode還是Datanode?
是NameNode,因為DataNode不夠可以很方便的水平拓展,而工作的NameNode只有一個,他的存儲能力完全取決于他的內存,所以。。。。,
但是其實NameNode一般不會成為瓶頸,因為一個塊記錄的元數(shù)據(jù)信息大小約為150B,如果每一個塊大小為128M的話,那么15G的NameNode內存可以存儲12PB的數(shù)據(jù)。
7:datanode明明已啟動,但是集群中的可用datanode列表中就是沒有,怎么辦?
已經不是處女,在她的Data目錄下,已經有其他NameNode的標記,這個NameNode不認。
8:文件下載到window中,為什么會報錯?
默認使用操作系統(tǒng)的內核進行磁盤數(shù)據(jù)的寫入,也就是需要一個winutil的工具,而默認的安裝包中不提供,所以需要編譯源碼或者設置為使用Java的進行磁盤寫入。
9:hadoop的HA(高可用)
MapReduce
1:MapReduce中,fileinputformat -> map -> shuffle -> reduce的過程
2:MapReduce中,job提交的過程
3:自定義Javabean作為數(shù)據(jù),需要extends writableandCompareble接口。
4:自定義outputformat,進行不同方向的處理。
5:MapReduce的一些應用場景
1、排序并且求 TOPOne 和TOPN
2、求某個用戶前幾個月的總流量,并且選擇出流量前幾名的用戶。
3、reduce端的join
4、map端join
5、求共同好友問題
hive
1:什么是hive?
一個將sql轉化為MapReduce程序的、單機版的、數(shù)據(jù)倉庫工具。通過關系型數(shù)據(jù)庫(mysql等)來記錄表元數(shù)據(jù)信息。真正的數(shù)據(jù)在HDFS中。
Hive利用HDFS存儲數(shù)據(jù),利用MapReduce查詢分析數(shù)據(jù)
hive2.0版本之后,都是基于Spark處理了。
安裝的時候,需要注意jline的版本沖突。
2:如何啟動?
3:執(zhí)行的sql的形式
hiveshell、 hive -e “sql命令”、 hive -f “一個包含著很多SQL語句的文件”
4:hive的創(chuàng)建表操作
內部表、外部表 就差連個關鍵字(external 和 location)
分區(qū)表、分桶表
5:hive查詢表
join
動態(tài)分區(qū)
分組查詢
復雜的那個累計報表操作。
6:hive自定義函數(shù)(UDF)
sqoop
利用hadoop的map端進行數(shù)據(jù)的并行導入導出。
安裝在HDFS上,配置HDFS的路徑和Hive路徑即可。
flume
1:agent:sources 、 channel 、 sinks
2:sources:exec、spooldir、arvo (加一個攔截器)
3:channel:men 、 disk
4:sinks:arvo 、HDFS、kafka
5:flume安裝在數(shù)據(jù)源這一邊。
6:如何自定義攔截器?
- class myiterceptor implements Iterceptor
- //里面有一個靜態(tài)的公共內部類。
- public static class mybuilder implements Iterceptor.Builder
7:如何實現(xiàn)flume的多級連接,以及如何實現(xiàn)高可用?
大數(shù)據(jù)實時storm部分
storm
1 : storm是一個實時的計算框架,只負責計算,不負責存儲。它通過spout的open和nextTuple方法去外部存儲系統(tǒng)(kafka)獲取數(shù)據(jù),然后傳送給后續(xù)的bolt處理,
bolt利用prepare和execute方法處理完成后,繼續(xù)往后續(xù)的bolt發(fā)送,或者根據(jù)輸出目錄,把信息寫到指定的外部存儲系統(tǒng)中。
2:storm的數(shù)據(jù)不丟失原理
交叉收到的數(shù)據(jù)做異或元算中間結果不為0的原理。
3:設置spout_max_pending (可以限流)
4:jstorm的通信機制,每一個:worker都有一個接受線程和輸出線程
5:storm的架構分析
nimbus、zookeeper、supervisor、worker
nimbus:接受任務請求,并且進行任務的分發(fā),最后寫入到zookeeper中。
supervisor:接受nimbus的任務調度,然后啟動和管理屬于自己的worker進程,supervisor是可以快速失敗的,不影響任務的執(zhí)行。
我們可以寫一個腳本來監(jiān)控supervisor的進程,如果不存在了,立馬啟動,就可以了。
worker:啟動spoutTask、boltTask等等任務,去執(zhí)行業(yè)務邏輯。
6:storm的編程模型
topology:由spout和bolt組成的一個流程圖。他描述著本次任務的信息
- spout:
- open
- nexttuple
- declareOutputFields
- bolt:
- prepare
- execute
- declareOutputFields
6:storm的tuple結構,它里面有兩個數(shù)據(jù)結構,一個list、一個是map
list:記錄著信息
map:記錄著每個字段對應的下表,通過找到下邊再去上面的list中找數(shù)據(jù)。
7:storm任務提交的過程
kafka
1、kafka和jms的區(qū)別
2、kafka的topic理解
topic是邏輯存在的,真正在物理磁盤中的體現(xiàn)是partitioner,一個topic可以對應多個partition,不同的paritition存放在不同的broker中,以提高并發(fā)存儲能力。
3、partitioner
partition是topic信息在屋里存儲中的具體體現(xiàn),在磁盤中它是一個文件夾,名字是topic名字_partition編號。4、segment
每個partition對對應多個segment文件,默認大小是1G,為了快速定位到指定的offset位置。
5、kafka為什么這么快
1/使用了操作系統(tǒng)使用的pagecache緩存,緩存大,緩存到一定量的數(shù)據(jù)時,以順序寫入的方 式寫入到磁盤中。
因為:磁盤順序寫入的方式非常的快=>600MB/s,而隨機存儲只有100kb/s左右。
2/使用操作系統(tǒng)的sendfile技術。在讀取信息發(fā)送的時候,不需要經過用戶區(qū),而是在os端直接發(fā)送,可以減少很多步驟。
6、為什么要多個partitioner7、為什么每個partitioner需要切分為多個segment文件
8、kafka的HA
對partitioner分區(qū)進行備份,利用zookeeper的選舉機制選擇leader。數(shù)據(jù)的生產存儲和消費讀取都是有l(wèi)eader負責,其他的replicatition只是負責備份而已。
9、kafka如何用shell腳本來講一個文件讀寫進去?10、kafka如何用JavaAPI實現(xiàn)生產者和消費者?
大數(shù)據(jù)一站式解決方案:Scala和Spark部分
scala回顧
1、如何定義變量
2、如何定義函數(shù)、方法,如何在將函數(shù)作為方法的參數(shù)傳入進去?
3、條件判斷語句,循環(huán)控制語句
4、集合操作:Array、list、set、tuple、map (注意:可變和不可變的區(qū)別)5、樣例類的使用6、trit、抽象類的使用7、主構造器和輔助構造器的使用
8、scala的高級特性
高階函數(shù):作為值得函數(shù)、匿名函數(shù)、閉包、柯里化
隱式轉換:一個類對象中,如果他沒有摸一個功能,但是我們有想要它實現(xiàn),可以使用英式轉換的方式。
- object MyPredef{
- //定義隱式轉換方法
- implicit def fileReadToRichFile(file: File)=new RichFile(file)
- }
使用:
import MyPredef._9、Actor
寫起來像多線程,用起來像socket10、akka
ActorSystem.actorOf()創(chuàng)建一個Actor,
創(chuàng)建的同時,就是執(zhí)行Actor中的prestart方法,去初始化一些信息。
Spark RDD
1、SparkRDD叫做:彈性分布式數(shù)據(jù)集,其實就是一個類,用來描述:任務的數(shù)據(jù)從哪里讀取、用那個算進行計算、得到的結果有存放在哪里、RDD之間的依賴關系是款以來還是窄依賴
2、RDD有五個特點
一系列分區(qū)
每個算子作用在每個分區(qū)上
一系列依賴關系
最有位置(如果從HDFS上讀取數(shù)據(jù))
3、RDD的兩種算子Transformation和Action
Transformation是懶加載,只是定義了這個算子的任務,該如何做,但是還沒有做。
Action是立即執(zhí)行,當執(zhí)行到Action時,會觸發(fā)DAGSchudle切分stage,切分完成后,有TaskScheduler將任務通過DriverActor發(fā)送到executor中執(zhí)行。
4、RDD的幾個復雜的Transformation
- ->combineByKey(x=>x,(a:List[String],b:String) => a :+ b,
- (m:List[String],n:List[String])=> m ++ n)
第一個參數(shù)表示分組后的第一個值如何處理,
第二個參數(shù)表示后續(xù)的值和前一個值如何處理,
第三個參數(shù)表示,map端處理完成后,在reduce端如何對這些list進行處理。
->aggregate(“初始量,可以是String也可以是int”)(第一個func,第二個func)
初始量作用于沒一個分區(qū),第一個func作用于map端,第二個func作用于reduce端。
->reduceByKey(_+_) 作用于map端和reduce端,可以進行局部聚合。
其實reduceByKey和aggregateByKey在底層都調用了combineByKey方法來實現(xiàn)響應的功能。
->mapPartitions
對每一個分區(qū)進行操作,直接在里面使用匿名函數(shù)即可
當然如果邏輯非常復雜也是可以考慮在外面先定義好這個函數(shù)之后在傳輸進去。
- rdd1.mapPartitions((it:Iterator[String]) => {
- it.toList.map(x => (x,1)).iterator
- })
- >mapPartitionsWithIndex
首先定義一個函數(shù),當然也可以寫在里面作為匿名函數(shù)
- val func = (index:Int, it:Iterator[Int]) => {
- it.toList.map(x => ("index:" + index, x)).iterator
- }
- rdd1.mapPartitionsWithIndex(func).collect
5、RDD自定義Partitioner
- //自定義分區(qū)器,重寫里面的getPartition方法和numPartitions方法。
- //構造這個對象的時候,就把所有情況的信息傳輸過來,然后在里面進行分類處理。
- class HostPartition(hostArr:Array[String]) extends Partitioner{
- //對所有的數(shù)據(jù)進行分類,每一種類型對應一個int編號。所以使用map比較合適。
- val map = new mutable.HashMap[String,Int]()
- for(index <- 0 until(hostArr.length)){
- map.put(hostArr(index),index)
- }
- //重寫getPartition的方法。
- override def getPartition(key: Any): Int = {
- map.getOrElse(key.toString,0)
- }
- override def numPartitions: Int = hostArr.length
- }
- 應用:
- val hostPartition: HostPartition = new HostPartition(hostList)
- val allPartitionRDD: RDD[(String, (String, Int))] = host_url_count.partitionBy(hostPartition)
6、自定義排序規(guī)則 ==>定義一個
- case class Gril(yanzhi:Int,nianling:Int) extends Ordered[Gril] with Serializable{
- override def compare(that: Gril): Int = {
- val yanzhiResult: Int = this.yanzhi.compareTo(that.yanzhi)
- if(yanzhiResult == 0){
- return this.nianling.compareTo(that.nianling)
- }
- return yanzhiResult
- }
- }
應用:
- val rdd2: RDD[(String, Int, Int)] = rdd1.sortBy(msg => Gril(msg._2,msg._3))
Spark的SQLContext
1、Spark整合Hive和HDFS 只需要將Hive的hive-site.xml ; hadoop的core-site.xml和hdfs-site.xml拷貝到Spark的conf目錄下即可。Spark就知道如何使用hive的表,同時也知道去哪個NameNode哪里都數(shù)據(jù)了。
2、DataFrame是什么?
是一個分布式數(shù)據(jù)集,對RDD的封裝。RDD有的方法他基本上都有
3、DataFrame如何創(chuàng)建?
三種方式:->RDD + case class
->RDD + structType
->sqlContext.read.format.options(Map())
4、DataFrame首先需要注冊成表結構之后才可以使用sqlContext來操作。
dF.registerTempTable(“person”)
5、使用sqlContext ==> 返回一個DataFrame
sqlContext.sql(“select * from person”)
6、DataFrame將數(shù)據(jù)寫入到HDFS或者mysql中
- val prop = new Properties()
- prop.put("user", "root")
- prop.put("password", "815325")
- //如果數(shù)據(jù)庫中沒有這個表,那么他也會創(chuàng)建一張表(很強大)
- resultDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/bigdata","result",prop)