分布式計(jì)算批處理引擎mapreduce第二部分

分布式計(jì)算:批處理引擎MapReduce(第二部分)張MapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4MapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4MapReduce編程模型

MapReduce將作業(yè)的整個(gè)運(yùn)行過程分為兩個(gè)階段：Map階段和Reduce階段

Map階段由一定數(shù)量的Map

Task組成輸入數(shù)據(jù)格式解析：InputFormat輸入數(shù)據(jù)處理：Mapper數(shù)據(jù)分組：Partitioner

Reduce階段由一定數(shù)量的Reduce

Task組成數(shù)據(jù)

拷貝數(shù)據(jù)按照key排序數(shù)據(jù)處理：Reducer數(shù)據(jù)輸出格式：OutputFormatMapReduce編程模型Split

0Split

1Split

2Split

3MapperRead(Inputformat)MapperRead(Inputformat)MapperRead(Inputformat)MapperRead(Inputformat)a1b1Partitionerc1c1Partitionera1c1Partitionerb1b1PartitionerShuffle

&

Sorta1a1Reducerb1b1Reducerb1c1c1Reducerc1Part-0Part-1Part-2Map階段Write(Outputformat)Write(Outputformat)Reduce階段Shuffle

&

SortShuffle

&

SortHDFSMapTaskReduceTaskWrite(Outputformat)inputsplitmapMapTaskreduceoutputReduceTaskinputsplitmapMapTaskbufferinmemorypartition,sort,combineand

spill

to

diskpartitionsmerge

ondiskmap

phasepartition

sortcombie

phaseinputsplitmapMapTaskreduceoutputReduceTaskother

reducersother

mappersbufferinmemorypartition,

sort,

combineand

spill

to

diskpartitionsmerge

ondiskmap

phasecopy

phasemerge

sort

phasereduce

phaseremotefetchmergemergemergememory

and

diskpartition

sortcombie

phase每一個(gè)Map

Task把輸出寫到內(nèi)存的bu?er中當(dāng)內(nèi)存bu?er到達(dá)一定閾值時(shí)，

線程會把數(shù)據(jù)寫到磁盤根據(jù)Par66oner，把數(shù)據(jù)寫入到不同的par66on對于每個(gè)par66on的數(shù)據(jù)進(jìn)行排序運(yùn)行Combiner隨著Map

Task的運(yùn)行，磁盤上的spill文件越來越多對于一個(gè)par66on下的spill分片，會進(jìn)行合并（把多個(gè)spill合并）運(yùn)行Combiner拷貝屬于它ReduceTask啟動后，從每個(gè)Map

Task上的par66on文件把這些文件歸并排序MapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4一批TB或者PB量級的文檔，需要完成以下功能：

搜索符合某種規(guī) 則表達(dá)式）的單詞或者句子；

統(tǒng)計(jì)相應(yīng)的單詞或者句子的數(shù)目；

按照數(shù)目對其進(jìn)行排序，并輸出最終結(jié)果。HDFS中的文檔shop[A-Z.]+shopA10排序shopB5shopC25shopA10shopB5shopC25shopD100shopD100頻率統(tǒng)計(jì)結(jié)果排序結(jié)果分為兩個(gè)作業(yè)：作業(yè)1：WordCount統(tǒng)計(jì)符合某種規(guī)則的單詞數(shù)目作業(yè)2：Sort按照單詞數(shù)目進(jìn)行全排序依賴于前一個(gè)作業(yè)的輸出結(jié)果Partitioner

默認(rèn)實(shí)現(xiàn)：hash(key)

modR

“hash(hostname(URL))

mod

R”Combiner

在每個(gè)mapper進(jìn)行局部規(guī)約job.setPartitionerClass(AgePartitioner.class);

/file/view/SamplePartitioner.javaMapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4與Linux管道機(jī)制一致通過標(biāo)準(zhǔn)輸入輸出實(shí)現(xiàn)進(jìn)程間通信標(biāo)準(zhǔn)輸入輸出是任何語言都有的幾個(gè)舉例：cat

1.txt

|

grep

“dong”

|

sortcat

1.txt

|

python

grep.py

|

java

sort.jarNodeManagerMapTaskorReduce

TaskStreaming

Codeinputkey/valuesoutputkey/valuesstd

instd

out

以標(biāo)準(zhǔn)輸入流作為輸入；C++：cinC：scanf

以標(biāo)準(zhǔn)輸出流作為輸出；C++：coutC：printf

可實(shí)現(xiàn)Mapper和Reducer，其他組件（InputFormat、Partitioner等需要用Java語言實(shí)現(xiàn)）

編譯程序，生成可執(zhí)行文件；g++

-o

mappermapper.cppg++

-o

reducerreducer.cpp

測試程序；cat

test.txt

|

./mapper

|

sort

|

./reducer

Hadoop上運(yùn)行Wordcount程序區(qū)分通用參數(shù)和命令行參數(shù)，通用參數(shù)應(yīng)放在命令行參數(shù)前面，否則不起作用。通用參數(shù)有7個(gè)：

-conf

-D

-fs

-jt

-files

-

libjars

-archives“-file”或者“-files”參數(shù)，設(shè)置要分發(fā)到各個(gè)節(jié)點(diǎn)上的文件，對于mapper和reducer文件，必須要用或者“-files”或“-file”指定。

-files

mapper,reducer

-file

mapper

-file

reducer每次運(yùn)行程序前，清空輸出bin/hadoop

fs

-rmr

/test/output測試mapper和reducercat

test.txt|

php

mapper.php

|

sort

|

php

reducer.php在Hadoop上運(yùn)行測試mapper和reducercat

test.txt|

sh

mapper.sh

|

sort

|

sh

reducer.sh在Hadoop上運(yùn)行

定義Hadoop

Counter

使用標(biāo)準(zhǔn)錯(cuò)誤輸出，格式為reporter:counter:<group>,<counter>,<amount>

在運(yùn)行過程中展示狀態(tài)信息

使用標(biāo)準(zhǔn)錯(cuò)誤輸出，格式為reporter:status:<message>

打印調(diào)試信息

使用標(biāo)準(zhǔn)錯(cuò)誤輸出，調(diào)試信息將被保存stderr文件中

獲取conf配置信息，比如reduce

task個(gè)數(shù)等

在環(huán)境變量中獲取，所有的“.”被替換成了“_”int

main(int

argc,

char

*argv[],

char

*env[])

{比如：“mapreduce.job.reduces”被換成了"mapreduce_job_reduces"

對多字段文本數(shù)據(jù)的支持

Java編程Hadoop最原始開發(fā)語言；支持所有功能，是其他編程方式的基礎(chǔ)。

Streaming編程僅用于開發(fā)Mapper和Reducer，其他組件需采用Java實(shí)現(xiàn)；天生支持文本格式，但二進(jìn)制格式支持較弱；通常用于簡單的文本數(shù)據(jù)處理，加快開發(fā)效率。MapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4JVM

HeapOff-heapmemory(JVM

overhead,

non-javaprocess)YARN

Containermapreduce.map.memory.mbmapreduce.map.memory.mbmapreduce.reduce.memory.mbmapreduce.map.java.optsmapreduce.reduce.java.opts-Dmapreduce.map.memory.mb=5120-Dmapreduce.reduce.memory.mb=5120-Dmapreduce.map.java.opts="-Xms4096m

-Xmx4096m"-Dmapreduce.reduce.java.opts="-Xms4096m

-Xmx4096m”mapreduce.map.java.optsmap.memory-map.java????mapreduce.map.maxattempts：map

task重試次數(shù)，默認(rèn)是4mapreduce.reduce.maxattempts：reduce

task重試次數(shù)，默認(rèn)是4mapreduce.map.failures.maxpercent：最大允許的map失敗率，默認(rèn)是0mapreduce.reduce.failures.maxpercent：最大允許的reduce失敗率，默認(rèn)是0jobMaptaskMaptaskMaptaskReducetask??mapreduce.map.speculative：是否打開map階段的推測執(zhí)行機(jī)制

mapreduce.reduce.speculative：是否打開reduce階段的推測執(zhí)行機(jī)制jobMaptaskMaptaskMaptask運(yùn)行時(shí)間：5min運(yùn)行時(shí)間：1min運(yùn)行時(shí)間：1min運(yùn)行時(shí)間：>5minmapreduce.input.fileinputformat.split.minsize:

每個(gè)map

task處理的數(shù)據(jù)量（大于一個(gè)block大小，以減少map

task數(shù)目）MaptaskMaptaskblockblockblockblockblockblockblock

blockHDFSMaptaskMaptaskMapReduce

On

YARNMapReduce

Java高級編程2MapReduce多語言編程3MapReduce編程模型回顧15

分布式日志分析系統(tǒng)：ETL模塊總結(jié)6MapReduce調(diào)優(yōu)小技巧4FlumeAppSiteDeviceSqoopHDFSHive(Meta

Store

Server)DatabaseTablePartitionsETLloadcollectloadPrestoSQLMRMysql/Oracle用戶信息產(chǎn)品信息用戶行為數(shù)據(jù)批處理線

ETLExtract-‐Transform-‐Load

將數(shù)據(jù)從來源端經(jīng)過抽?。╡xtract）、轉(zhuǎn)換（transform）、加載（load）至目的端的過程

數(shù)據(jù)與轉(zhuǎn)換不符合要求的數(shù)據(jù)主要是有不完整的數(shù)據(jù)、錯(cuò)誤的數(shù)據(jù)、重復(fù)的數(shù)據(jù)不一致的數(shù)據(jù)轉(zhuǎn)換、數(shù)據(jù)粒度的轉(zhuǎn)換，以及一些商務(wù)規(guī)則的計(jì)算

使用Hadoop

Streaming抽取日志部分字段轉(zhuǎn)換日期字段的日期格式HADOOP_CMD="hadoop

jar

/ho