大數(shù)據(jù)及MapReduce編程模型94.ppt

1、Hust_ 2014年 8 月,大數(shù)據(jù)及其并行編程模型概述,主 要 內(nèi) 容,一、大數(shù)據(jù)概述 二、應對大數(shù)據(jù)的系統(tǒng)思維 三、MapReduce 并行編程詳解,2,注：本課件前30頁PPT來源于國防科大 李東升教授：“大數(shù)據(jù)時代的挑戰(zhàn)和探索”,互聯(lián)網(wǎng)應用數(shù)據(jù)急劇增長 互聯(lián)網(wǎng)用戶數(shù)量巨大，日益活躍 微博、論壇、電子商務網(wǎng)站等 互聯(lián)網(wǎng)上的用戶生成數(shù)據(jù)（User Generated Content， UGC） 淘寶網(wǎng)每天新增數(shù)據(jù)40TB以上 百度每天處理10PB量級的數(shù)據(jù)， 總數(shù)據(jù)量達1000PB,應用背景,注：本課件前30頁PPT來源于國防科大 李東升教授：“大數(shù)據(jù)時代的挑戰(zhàn)

2、和探索”,一、大數(shù)據(jù)概述, 隨著信息化的推進，國民經(jīng)濟、國家安全 等領域的數(shù)據(jù)不斷增長 物聯(lián)網(wǎng)、移動通信電話、手機短信、語音數(shù)據(jù) 遙感、公共安全、醫(yī)療、交通、情報等很多領域 高分辨率衛(wèi)星（影像）、城市監(jiān)控攝像頭（視頻）、 據(jù)報道，武漢監(jiān)控攝像頭已超過25萬個，如采用1080P高清攝 像頭（一天產(chǎn)生數(shù)據(jù)量40GB以上），整個城市每天新增監(jiān)控 數(shù)據(jù)10PB以上,應用背景, 科學實驗數(shù)據(jù)規(guī)模巨大，增長迅猛 生物工程 氣候監(jiān)測 高能物理 天文觀測 生態(tài)環(huán)境 .,氣候研究,華大基因測序目 前每天產(chǎn)生數(shù)據(jù) 約15TB，一年超 過5PB一,歐洲CERN對撞 機每年產(chǎn)生的數(shù) 據(jù)量超過15 PB,基因測序,應用

3、背景,全 球 數(shù) 據(jù) 量, IDC報告預測：未來 十年，全球數(shù)據(jù)量繼 續(xù)迅速增長,Amount of digital information created and replicated in a year, ,年均增長率超過40% 2009年0.8ZB 2020年35ZB 1 ZB 106PB 月球,容量4GB的DVD光 用容量4GB的DVD光盤存儲，DVD可從地球排至月球,G-T-P-E-Z-Y, 維基（Wiki）百科的定義 Big data is a collection of data sets so large and complex that it becomes difficul

4、t to process using on-hand database management tools IDC的定義 Big data technologies describe a new generation of technologies and architectures, designed to economically extract value from very large volumes of a wide variety of data, by enabling high-velocity capture, discovery, and/or analysis.,什么是大

5、數(shù)據(jù),大數(shù)據(jù)是超大、復雜的數(shù)據(jù)集，現(xiàn)有的數(shù)據(jù)庫管理技術難以應對,大數(shù)據(jù)技術描述了新一代的技術和架構，通過高速的數(shù)據(jù)獲取、發(fā)現(xiàn)和分析技術，以經(jīng)濟的方式從各種超大規(guī)模的數(shù)據(jù)中提取價值,一、大數(shù)據(jù)概述, Volume：規(guī)模大 從PB級到ZB級 1 ZB 106* PB Variety：多樣化 結構化、非結構化 文本、圖像、視頻等 Velocity：變化快 批處理/離線數(shù)據(jù)、流/實時/在線數(shù)據(jù)等 Value/ Veracity：價值稀疏 /數(shù)據(jù)質(zhì)量 噪音和無用信息很多,一、大數(shù)據(jù)概述,大數(shù)據(jù)的特點, 大數(shù)據(jù)技術對經(jīng)濟社會和科研都在產(chǎn)生重 要影響 互聯(lián)網(wǎng)產(chǎn)業(yè)、電子商務推薦、日常生活,大數(shù)據(jù)的影響,季節(jié)性

6、流感是一個重要的公共衛(wèi)生問題：WHO估計， 全球每年25萬至50萬人因此死亡 及時監(jiān)測到疾病的傳播情況，盡快采取應對措施 2008年，Google通過處理網(wǎng)絡搜索日志中的幾千億查詢數(shù)據(jù)，訓練建立流感疾病監(jiān)測的數(shù)學模型，比美國病控制和預防中心提前1-2周給出流感的傳播情況 論文發(fā)表在Nature (2009.2): Detecting Influenza Epidemics using Search Engine Query Data, 大數(shù)據(jù)技術對經(jīng)濟社會和科研都在產(chǎn)生重 要影響 科學研究,三種科研模式：理論、實驗、計算 第四模式：數(shù)據(jù)密集型的科學發(fā)現(xiàn) 圖靈獎獲得者Jim Gray 2007年

7、提出 專輯：Nature (2008.9): ”Big Data”, Science (2011.2): ”Dealing with data”,大數(shù)據(jù)的影響, 2012年3月29日，美國政府宣布投資2億 美元啟動“大數(shù)據(jù)研發(fā)計劃” （ Big Data R reduce(String output_key, Iterator intermediate_values): / output_key: a word / output_values: a list of counts int result = 0; for each v in intermediate_values: result

8、+= ParseInt(v); Emit(AsString(result);,其他示例,分布式檢索 map函數(shù)挑選出滿足特定模式的行，并將其組裝成元組輸出。 reduce函數(shù)是一個簡單的確認函數(shù)，它完成的工作僅僅是將中間元組拷貝到輸出中。 計算URL訪問頻率 map函數(shù)處理web網(wǎng)頁的訪問日志，并輸出。 reduce函數(shù)將每個URL的訪問次數(shù)加起來，輸出,其他示例,翻轉(zhuǎn)web-link圖 在每個作為源的頁面中，檢查其連接URL，并逐個輸出元組。 reduce函數(shù)將連接到每個target的所有source組合起來，形成list列表，輸出 每個站點的術語向量 術語向量表示出在一篇文章中或者一組文章

9、中最重要的單詞，通常以元組的方式。 map函數(shù)輸出每個文章的 （hostname通過文章的URL分析得到）。 reduce函數(shù)取出不常用的術語，將其余的相加，得到最終的對,其他示例,倒排索引 map函數(shù)分析每個文檔,然后產(chǎn)生一個(詞,文檔號)對的序列.reduce函數(shù)接受一個給定詞的所有對,排序相應的文檔IDs,并且產(chǎn)生一個(詞,文檔ID列表)對.所有的輸出對集形成一個簡單的倒排索引 分布式排序 map函數(shù)從每個記錄提取key,并且產(chǎn)生一個(key,record)對.reduce函數(shù)不改變?nèi)魏蔚膶?,“實踐是檢驗真理的唯一標準”,實踐證明，MapReduce是出色的分布式計算模型 Google

10、宣布，其對分布于1000臺計算機上的1TB數(shù)據(jù)進行排序僅僅需要68s 對4000臺計算機上的1PB數(shù)據(jù)進行排序處理僅需要6小時2分鐘（每次測試至少會損壞1塊硬盤） 在08年1月份，Google MapReduce平均每天的數(shù)據(jù)處理量是20PB，相當于美國國會圖書館當年5月份存檔網(wǎng)絡數(shù)據(jù)的240倍,Hadoop上的MapReduce,job: 是客戶端程序想要完成的一系列工作的集合。包括輸入數(shù)據(jù)，MapReduce程序和配置信息。 task: Hadoop將job分解為tasks 有兩種類型的task: map task和reduce task jobtracker和tasktracker: 用

11、來控制job執(zhí)行的 tasktracker運行task，并向jobtracker報告進度信息 jobtracker記錄下每一個job的進度信息，如果一個task失敗，jobtracker會將其重新調(diào)度到另外的tasktracker上。,Hadoop-MapReduce工作原理,Hadoop Streaming and Pipes,Hadoop流 允許用Java以外的語言來編寫Map和Reduce函數(shù) Hadoop管道 C+接口,流和管道及子進程的關系,進度和狀態(tài)更新,Hadoop-MapReduce,Mapper public static class *Mapper extends Mapp

12、er Reducer public static class *Reducer extends educer Driver Package org.apache.hadoop.mapreduce.Job; org.apache.hadoop.mapreduce.Mapper; org.apache.hadoop.mapreduce.Reducer;,接口描述,public interface Mapper extends JobConfigurable, Closeable void map(K1 key, V1 value, OutputCollector output, Reporter

13、reporter) throws IOException; public interface Reducer extends JobConfigurable, Closeable void reduce(K2 key, Iterator values, OutputCollector output, Reporter reporter) throws IOException; ,Mapper,public static class Map extends Mapper private final static IntWritable one = new IntWritable(1); priv

14、ate Text word = new Text(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException String line = value.toString(); StringTokenizer tokenizer = new StringTokenizer(line); while (tokenizer.hasMoreTokens() word.set(tokenizer.nextToken(); context.write(wo

15、rd, one); ,Reducer,public static class Reduce extends Reducer public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException int sum = 0; for (IntWritable val : values) sum += val.get(); context.write(key, new IntWritable(sum); ,Driver,public static void main

16、(String args) throws Exception Configuration conf = new Configuration(); Job job = new Job(conf, wordcount); job.setJarByClass(WordCount.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); job.setInp

17、utFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.addInputPath(job, new Path(args0); FileOutputFormat.setOutputPath(job, new Path(args1); job.waitForCompletion(true); ,Input Files,輸入文件一般 保存在HDFS中 文件的類型不固定，可能是文本的，也有可能是其它形式的文件 文件經(jīng)常很大，甚至有幾十個GB Input

18、會被分成input split，split由record 組成。map處理每一個record，并且返回key和value的對 MapReduce程序并不需要直接處理InputSplit，由InputFormat創(chuàng)建的,InputSplits,InputSplit定義了輸入到單個Map任務的輸入數(shù)據(jù) InputSplit將文件分為64MB的大小 hadoop-site.xml中的mapred.min.split.size參數(shù)控制這個大小 mapred.tasktracker.map.taks.maximum用來控制某一個節(jié)點上所有map任務的最大數(shù)目,RecordReader,InputSpli

19、t定義了一項工作的大小，但是沒有定義如何讀取數(shù)據(jù) RecordReader實際上定義了如何從數(shù)據(jù)上轉(zhuǎn)化為一個(key,value)對，從而輸出到Mapper類中 TextInputFormat提供了LineRecordReader,InputFormat,定義了這些文件如何分割，讀取 InputFile提供了以下一些功能 選擇文件或者其它對象，用來作為輸入 定義InputSplits，將一個文件分開成為任務 為RecordReader提供一個工廠，用來讀取這個文件 有一個抽象的類FileInputFormat，所有的輸入格式類都從這個類繼承這個類的功能以及特性。當啟動一個Hadoop任務的時候

20、，一個輸入文件所在的目錄被輸入到FileInputFormat對象中。FileInputFormat從這個目錄中讀取所有文件。然后FileInputFormat將這些文件分割為一個或者多個InputSplits。 通過在JobConf對象上設置JobConf.setInputFormat設置文件輸入的格式,預定義的文件輸入格式,各種InputFormat,TextInputFormat，默認的格式，每一行是一個單獨的記錄，并且作為value，文件的偏移值作為key KeyValueInputFormat，這個格式每一行也是一個單獨的記錄，但是Key和Value用Tab隔開，是默認的Output

21、Format，可以作為中間結果，作為下一步MapReduce的輸入。 SequenceFileInputFormat 基于塊進行壓縮的格式 對于幾種類型數(shù)據(jù)的序列化和反序列化操作 用來將數(shù)據(jù)快速讀取到Mapper類中,Writable接口,Hadoop使用Writable做序列化 定義了兩個方法 二進制寫入DataOutput流 二進制讀取DataInput流 Hadoop自帶一系列Writable實現(xiàn)，可以滿足絕大多數(shù)需要 可以自定義Writable，控制二進制表示和排序,實現(xiàn)Writable接口的例子,publicclassMyWritableimplementsWritable /Som

22、edata privateintcounter; privatelongtimestamp; publicvoidwrite(DataOutputout)throwsIOException out.writeInt(counter); out.writeLong(timestamp); publicvoidreadFields(DataInputin)throwsIOException counter=in.readInt(); timestamp=in.readLong(); publicstaticMyWritableread(DataInputin)throwsIOException M

23、yWritablew=newMyWritable(); w.readFields(in); returnw; ,Writable的Java基本封裝,Mapper,每一個Mapper類的實例生成了一個Java進程（在某一個InputSplit上執(zhí)行） 有兩個額外的參數(shù)OutputCollector以及Reporter，前者用來收集中間結果，后者用來獲得環(huán)境參數(shù)以及設置當前執(zhí)行的狀態(tài)。 現(xiàn)在用Mapper.Context提供給每一個Mapper函數(shù)，用來提供上面兩個對象的功能 數(shù)據(jù)壓縮,Partition 是在本地執(zhí)行的一個Reducer，滿足一定的條件才能夠執(zhí)行。,Sort,傳輸?shù)矫恳粋€節(jié)點上的

24、所有的Reduce函數(shù)接收到得Key,value對會被Hadoop自動排序（即Map生成的結果傳送到某一個節(jié)點的時候，會被自動排序）,Reduce,做用戶定義的Reduce操作 接收到一個OutputCollector的類作為輸出,OutputFormat,寫入到HDFS的所有OutputFormat都繼承自FileOutputFormat 每一個Reducer都寫一個文件到一個共同的輸出目錄，文件名是part-nnnnn，其中nnnnn是與每一個reducer相關的一個號（partition id） JobConf.setOutputFormat() RecordWriter 用來指導如何輸

25、出一個記錄到文件中,Output Format,容錯,由Hadoop系統(tǒng)自己解決 主要方法是將失敗的任務進行再次執(zhí)行 TaskTracker會把狀態(tài)信息匯報給JobTracker，最終由JobTracker決定重新執(zhí)行哪一個任務 為了加快執(zhí)行的速度，Hadoop也會自動重復執(zhí)行同一個任務，以最先執(zhí)行成功的為準 mapred.map.tasks.speculative.execution mapred.reduce.tasks.speculative.execution,調(diào)優(yōu),部分屬性除了配置文件之外還可以在MapReduce作業(yè)中動態(tài)修改 在MapReduce執(zhí)行過程中，特別是Shuffle階

26、段，盡量使用內(nèi)存緩沖區(qū)存儲數(shù)據(jù)，減少磁盤溢寫次數(shù)；同時在作業(yè)執(zhí)行過程中增加并行度，都能夠顯著提高系統(tǒng)性能，這也是配置優(yōu)化的一個重要依據(jù)。 由于每個Hadoop集群的機器和硬件之間都存在一定差別，所以Hadoop框架應根據(jù)其集群特性做配置優(yōu)化,IO屬性優(yōu)化,主要包括在Shuffle階段中相關的I/O過程的屬性 io.sort.factor屬性 int類型，Map端和Reduce端使用 該屬性設置在Map端和Reduce端都使用到的對文件Sort時一次合并的最大流，其默認值是10，即一次合并10個流。 在集群中，將其適當增大能夠提高并行度以縮短合并所需時間。將此默認值增加到100是比較常見的。 io.sort.mb屬性 int類型，Map端使用，Map輸出進行排序時使用的環(huán)形內(nèi)存緩沖區(qū)的大小，以M字節(jié)為單位，默認是100M。如果允許，應該增加它的值來減少磁盤溢寫的次數(shù)以提高性能。 io.sort.record.percent屬性 float類型，M