科多大數(shù)據(jù)小課堂之大數(shù)據(jù)分析之技術框架整理。

　　大數(shù)據(jù)離線部分

　　HDFS

　　1：HDFS的架構部分及工作原理

　　NameNode：負責管理元素據(jù)，將信息保存在內(nèi)存中

　　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只有一個，他的存儲能力完全取決于他的內(nèi)存。

　　但是其實NameNode一般不會成為瓶頸，因為一個塊記錄的元數(shù)據(jù)信息大小約為150B，如果每一個塊大小為128M的話，那么15G的NameNode內(nèi)存可以存儲12PB的數(shù)據(jù)。

　　7：datanode明明已啟動，但是集群中的可用datanode列表中就是沒有，怎么辦?

　　在她的Data目錄下，已經(jīng)有其他NameNode的標記，這個NameNode不認。

　　8：文件下載到window中，為什么會報錯?

　　默認使用操作系統(tǒng)的內(nèi)核進行磁盤數(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)建表操作

　　內(nèi)部表、外部表 就差連個關鍵字(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)的公共內(nè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的任務調(diào)度，然后啟動和管理屬于自己的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ù)時，以順序?qū)懭氲姆?式寫入到磁盤中。

　　因為：磁盤順序?qū)懭氲姆绞椒浅５目?>600MB/s,而隨機存儲只有100kb/s左右。

　　2/使用操作系統(tǒng)的sendfile技術。在讀取信息發(fā)送的時候，不需要經(jīng)過用戶區(qū)，而是在os端直接發(fā)送，可以減少很多步驟。

　　6、為什么要多個partitioner7、為什么每個partitioner需要切分為多個segment文件

　　8、kafka的HA

　　對partitioner分區(qū)進行備份，利用zookeeper的選舉機制選擇leader。數(shù)據(jù)的生產(chǎn)存儲和消費讀取都是有l(wèi)eader負責，其他的replicatition只是負責備份而已。

　　9、kafka如何用shell腳本來講一個文件讀寫進去?10、kafka如何用JavaAPI實現(xiàn)生產(chǎ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在底層都調(diào)用了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

　　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)