day08 MapReduce
生活随笔
收集整理的這篇文章主要介紹了
day08 MapReduce
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
PS：?HDFS對于MapReduce來說，HDFS就是一個就是一個客戶端。
PS:?離線就是?寫sql,sparkh還是寫sql
1. MAPREDUCE原理篇（1）
Mapreduce是一個分布式運算程序的編程框架，是用戶開發(fā)“基于hadoop的數(shù)據分析應用”的核心框架；
Mapreduce核心功能是將用戶編寫的業(yè)務邏輯代碼和自帶默認組件整合成一個完整的分布式運算程序，并發(fā)運行在一個hadoop集群上；
?
PS: 上圖為mapreduce的設計思想，以wordcount為例子，首先把一個任務分我 兩個階段。 第一map階段負責業(yè)務邏輯，第二數(shù)據的合并（比如按a-h合并，然后第二個安裝i-k）。中間還有還多負責的業(yè)務管理邏輯，由mr application master來管理，協(xié)調調度。
PS: MapReduce程序都是依賴于HDFS，以流的形式讀取
----------------------------------------------------------------------------------------------------------------
1.1 為什么要MAPREDUCE
（1）海量數(shù)據在單機上處理因為硬件資源限制，無法勝任
（2）而一旦將單機版程序擴展到集群來分布式運行，將極大增加程序的復雜度和開發(fā)難度
（3）引入mapreduce框架后，開發(fā)人員可以將絕大部分工作集中在業(yè)務邏輯的開發(fā)上，而將分布式計算中的復雜性交由框架來處理
-----------------------------------------------------------
設想一個海量數(shù)據場景下的wordcount需求：
| 單機版：內存受限，磁盤受限，運算能力受限 分布式： 1、文件分布式存儲（HDFS） 2、運算邏輯需要至少分成2個階段（一個階段獨立并發(fā)，一個階段匯聚） 3、運算程序如何分發(fā) 4、程序如何分配運算任務（切片） 5、兩階段的程序如何啟動？如何協(xié)調？ 6、整個程序運行過程中的監(jiān)控？容錯？重試？ |
?
可見在程序由單機版擴成分布式時，會引入大量的復雜工作。為了提高開發(fā)效率，可以將分布式程序中的公共功能封裝成框架，讓開發(fā)人員可以將精力集中于業(yè)務邏輯。
而mapreduce就是這樣一個分布式程序的通用框架，其應對以上問題的整體結構如下：
| 1、MRAppMaster(mapreduce application master) 2、MapTask 3、ReduceTask |
?
1.2 MAPREDUCE框架結構及核心運行機制
1.2.1 結構
一個完整的mapreduce程序在分布式運行時有三類實例進程：
1、MRAppMaster：負責整個程序的過程調度及狀態(tài)協(xié)調
2、mapTask：負責map階段的整個數(shù)據處理流程
3、ReduceTask：負責reduce階段的整個數(shù)據處理流程
-----------------------------------------------------------------------------
PS:寫程序代碼
1.配置項目，common和hdfs如之前配置
2.配置mapreduce
3.配置yarn。PS:
yarn Apache Hadoop YARN （Yet Another Resource Negotiator，另一種資源協(xié)調者）是一種新的 Hadoop 資源管理器，它是一個通用資源管理系統(tǒng)，可為上層應用提供統(tǒng)一的資源管理和調度，它的引入為集群在利用率、資源統(tǒng)一管理和數(shù)據共享等方面帶來了巨大好處。
------------------
YARN的基本思想是將JobTracker的兩個主要功能（資源管理和作業(yè)調度/監(jiān)控）分離，主要方法是創(chuàng)建一個全局的ResourceManager（RM）
和若干個針對應用程序的ApplicationMaster（AM）。這里的應用程序是指傳統(tǒng)的MapReduce作業(yè)或作業(yè)的DAG（有向無環(huán)圖）。
?
-------------------------------------------------------------------
package cn.itcast.bigdata.mr.wcdemo;import java.io.IOException;import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper;/*** KEYIN: 默認情況下，是mr框架所讀到的一行文本的起始偏移量，Long,* 但是在hadoop中有自己的更精簡的序列化接口，所以不直接用Long，而用LongWritable* * VALUEIN:默認情況下，是mr框架所讀到的一行文本的內容，String，同上，用Text* * KEYOUT：是用戶自定義邏輯處理完成之后輸出數(shù)據中的key，在此處是單詞，String，同上，用Text* VALUEOUT：是用戶自定義邏輯處理完成之后輸出數(shù)據中的value，在此處是單詞次數(shù)，Integer，同上，用IntWritable* * @author**/public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{/*** map階段的業(yè)務邏輯就寫在自定義的map()方法中* maptask會對每一行輸入數(shù)據調用一次我們自定義的map()方法*/@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//將maptask傳給我們的文本內容先轉換成StringString line = value.toString();//根據空格將這一行切分成單詞String[] words = line.split(" ");//將單詞輸出為<單詞，1>for(String word:words){//將單詞作為key，將次數(shù)1作為value，以便于后續(xù)的數(shù)據分發(fā)，可以根據單詞分發(fā)，以便于相同單詞會到相同的reduce taskcontext.write(new Text(word), new IntWritable(1));}}} package cn.itcast.bigdata.mr.wcdemo;import java.io.IOException;import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer;/*** KEYIN, VALUEIN 對應 mapper輸出的KEYOUT,VALUEOUT類型對應* * KEYOUT, VALUEOUT 是自定義reduce邏輯處理結果的輸出數(shù)據類型* KEYOUT是單詞* VLAUEOUT是總次數(shù)* @author**/ public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{/* 到這里的時候，數(shù)據已經按照一定的特性（規(guī)律如 統(tǒng)計a-h） 分組好了，所以入參的時候是一個同樣的key，values是Itrater專門用來迭代累加； 然后再執(zhí)行第二組 hello ,1,...指導統(tǒng)計完* <angelababy,1><angelababy,1><angelababy,1><angelababy,1><angelababy,1>* <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>* <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1>* 入參key，是一組相同單詞kv對的key*/@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int count=0;/*Iterator<IntWritable> iterator = values.iterator();while(iterator.hasNext()){count += iterator.next().get();}*/for(IntWritable value:values){count += value.get();}context.write(key, new IntWritable(count));}} package cn.itcast.bigdata.mr.wcdemo;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;/*** 相當于一個yarn集群的客戶端* 需要在此封裝我們的mr程序的相關運行參數(shù)，指定jar包* 最后提交給yarn* @author**/ public class WordcountDriver {public static void main(String[] args) throws Exception {if (args == null || args.length == 0) {args = new String[2];args[0] = "hdfs://192.168.8.10:9000/wordcount/input/wordcount.txt";args[1] = "hdfs://192.168.8.10:9000/wordcount/output";}Configuration conf = new Configuration();//設置的沒有用! ?????? 因為在linux下允許，所以這些注釋了 // conf.set("HADOOP_USER_NAME", "hadoop"); // conf.set("dfs.permissions.enabled", "false");// ---因為linux和windows操作系統(tǒng)的環(huán)境變量不同，所以不能直接運行在windows上，現(xiàn)在打包到linux系統(tǒng)上，那么，就不用注釋了（因為集群上已經配置過了）/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路徑job.setJarByClass(WordcountDriver.class);//指定本業(yè)務job要使用的mapper/Reducer業(yè)務類job.setMapperClass(WordcountMapper.class);job.setReducerClass(WordcountReducer.class);//指定mapper輸出數(shù)據的kv類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//指定最終輸出的數(shù)據的kv類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//指定job的輸入原始文件所在目錄FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的輸出結果所在目錄FileOutputFormat.setOutputPath(job, new Path(args[1]));//將job中配置的相關參數(shù)，以及job所用的java類所在的jar包，提交給yarn去運行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);} }
PS:打包文件
?hadoop jar wordcount.jar cn.itcast.bigdata.mr.wcdemo.WordcountDriver /wordcount/input /wordcount/output1
PS:?紅色命令? 其實就是?調用?jar命令 ，只是把關聯(lián)的jar鏈接上了
?PS:在界面端也有一個專門界面。
PS:這個文件可以part-r-0000可以自定義文件的個數(shù)，現(xiàn)在把所有的文件都放在part-r-0000中。
PS：程序先啟動的時候先啟動Mr application master,然后再啟動map task 、reducetask
?
------------------------------------------------------------------------------------WordCount程序運行流程分析------------------------------------------------
?
PS:上圖應該這么看，首先待處理文件。當submit（）的時候，會看到文件的規(guī)模，進行map()劃分（啟動maptask的多少，這個文件是job.split），然后job.split、wc.jar、job。xml提交到云端的YARN進行管理。然后yarn啟動mr appmaster，啟動相應的Node Manager進行管理。
執(zhí)行map（）任務，他是通過InputFormat組件以流的形式讀入一定范圍的數(shù)據進來，每一個任務執(zhí)行完了，接著通過另一個組件outputController會生成分區(qū)且排序的結果文件，
再進入reducetask()任務， 每傳過來maptask可以根據分區(qū)號進行分區(qū)，也就是輸出的part-r-0000X的變化的值。
在mapreduce線程中，執(zhí)行業(yè)務邏輯，通過outputFormat輸出，最后生成文件（讀取多個的文件，輸出會進行文件）。
PS:? 上圖為輸出的應用結果，? ?key為單詞字符，value是每個單詞的個數(shù)
?---------------------------------
?
?-----------------------------------------MapReduce?應用編譯執(zhí)行
?
1. PS:start-dfs.sh //啟動應用
start-yarn.sh //啟動yarn管理器
?--------------------------------------------另一個程序練習
/*
PS： 打印什么內容是 由key 和 輸出 value對象的toString方法所決定的。
*/
?
?
PS:文件在hdfs上，用jar包進行編譯執(zhí)行。?
?
--------------------------切片和并行度的概念------------
1.3.1 mapTask并行度的決定機制
一個job的map階段并行度由客戶端在提交job時決定
而客戶端對map階段并行度的規(guī)劃的基本邏輯為：
將待處理數(shù)據執(zhí)行邏輯切片（即按照一個特定切片大小，將待處理數(shù)據劃分成邏輯上的多個split），然后每一個split分配一個mapTask并行實例處理
Ps:由HDFS讀取到 日志文件后，根據文件block的大小進行并行度的確定。通常大文件直接有文件塊決定，決定時機是在 waitForCompletion這個函數(shù)執(zhí)行時，會生成一個分塊的文件。然后再mapreduce執(zhí)行時，MRAppMaster就可以讀取到信息進行調度操作了。
PS:上圖為提交數(shù)據的完成的信息，里面有包含分塊的信息。所有配置參數(shù)，這些參數(shù)可以讓MRAppMaster讀取數(shù)據
?PS :客戶端提交ＭＲ程序job的流程
?
?-------------------------------------------------------------------------------------
按照不同號碼分區(qū)
package cn.itcast.bigdata.mr.provinceflow;import java.util.HashMap;import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner;/*** K2 V2 對應的是map輸出kv的類型， * @author* 返回的是分區(qū)地址的hashcode*/ public class ProvincePartitioner extends Partitioner<Text, FlowBean>{public static HashMap<String, Integer> proviceDict = new HashMap<String, Integer>();static{proviceDict.put("136", 0);proviceDict.put("137", 1);proviceDict.put("138", 2);proviceDict.put("139", 3);}@Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {String prefix = key.toString().substring(0, 3);Integer provinceId = proviceDict.get(prefix);return provinceId==null?4:provinceId;} } package cn.itcast.bigdata.mr.provinceflow;import java.io.IOException;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class FlowCount {static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString(); //將一行內容轉成stringString[] fields = line.split("\t"); //切分字段String phoneNbr = fields[1]; //取出手機號long upFlow = Long.parseLong(fields[fields.length-3]); //取出上行流量下行流量long dFlow = Long.parseLong(fields[fields.length-2]);context.write(new Text(phoneNbr), new FlowBean(upFlow, dFlow));}}static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>{//<183323,bean1><183323,bean2><183323,bean3><183323,bean4>....... @Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {long sum_upFlow = 0;long sum_dFlow = 0;//遍歷所有bean，將其中的上行流量，下行流量分別累加for(FlowBean bean: values){sum_upFlow += bean.getUpFlow();sum_dFlow += bean.getdFlow();}FlowBean resultBean = new FlowBean(sum_upFlow, sum_dFlow);context.write(key, resultBean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路徑job.setJarByClass(FlowCount.class);//指定本業(yè)務job要使用的mapper/Reducer業(yè)務類job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);//指定我們自定義的數(shù)據分區(qū)器job.setPartitionerClass(ProvincePartitioner.class);//同時指定相應“分區(qū)”數(shù)量的reducetaskjob.setNumReduceTasks(5); //數(shù)量必須和分區(qū)相對應，可以為1 可以大于分區(qū)個數(shù)。單數(shù)少于1到分區(qū)數(shù)會報錯。//指定mapper輸出數(shù)據的kv類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//指定最終輸出的數(shù)據的kv類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//指定job的輸入原始文件所在目錄FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的輸出結果所在目錄FileOutputFormat.setOutputPath(job, new Path(args[1]));//將job中配置的相關參數(shù)，以及job所用的java類所在的jar包，提交給yarn去運行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);}}----------------------------作業(yè)：對以下數(shù)據進行排序，按照流量排序，這個是在之前生成數(shù)據的基礎上做的
PS:不要想著程序一下子全部完成，也可以分步驟解決問題、
package cn.itcast.bigdata.mr.flowsum;import java.io.DataInput; import java.io.DataOutput; import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class FlowBean implements WritableComparable<FlowBean>{private long upFlow;private long dFlow;private long sumFlow;//反序列化時，需要反射調用空參構造函數(shù)，所以要顯示定義一個public FlowBean(){}public FlowBean(long upFlow, long dFlow) {this.upFlow = upFlow;this.dFlow = dFlow;this.sumFlow = upFlow + dFlow;}public void set(long upFlow, long dFlow) {this.upFlow = upFlow;this.dFlow = dFlow;this.sumFlow = upFlow + dFlow;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getdFlow() {return dFlow;}public void setdFlow(long dFlow) {this.dFlow = dFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow = sumFlow;}/*** 序列化方法*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(dFlow);out.writeLong(sumFlow);}/*** 反序列化方法* 注意：反序列化的順序跟序列化的順序完全一致*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();dFlow = in.readLong();sumFlow = in.readLong();}@Overridepublic String toString() { return upFlow + "\t" + dFlow + "\t" + sumFlow;}@Overridepublic int compareTo(FlowBean o) {return this.sumFlow>o.getSumFlow()?-1:1; //從大到小, 當前對象和要比較的對象比, 如果當前對象大, 返回-1, 交換他們的位置(自己的理解) }} package cn.itcast.bigdata.mr.flowsum;import java.io.IOException;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import cn.itcast.bigdata.mr.flowsum.FlowCount.FlowCountMapper; import cn.itcast.bigdata.mr.flowsum.FlowCount.FlowCountReducer;/*** 13480253104 180 180 360 13502468823 7335 110349 117684 13560436666 1116 954* 2070* * @author* */ public class FlowCountSort {static class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {FlowBean bean = new FlowBean();//！！！雖然這是一個對象，但是是序列化，所以不用擔心數(shù)據操作Text v = new Text();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 拿到的是上一個統(tǒng)計程序的輸出結果，已經是各手機號的總流量信息String line = value.toString();String[] fields = line.split("\t");String phoneNbr = fields[0];long upFlow = Long.parseLong(fields[1]);long dFlow = Long.parseLong(fields[2]);bean.set(upFlow, dFlow);v.set(phoneNbr);context.write(bean, v);}}/*** 根據key來掉, 傳過來的是對象, 每個對象都是不一樣的, 所以每個對象都調用一次reduce方法* @author: 張政* @date: 2016年4月11日 下午7:08:18* @package_name: day07.sample*/static class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {// <bean(),phonenbr> @Overrideprotected void reduce(FlowBean bean, Iterable<Text> values, Context context) throws IOException, InterruptedException {context.write(values.iterator().next(), bean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路徑job.setJarByClass(FlowCountSort.class);//指定本業(yè)務job要使用的mapper/Reducer業(yè)務類job.setMapperClass(FlowCountSortMapper.class);job.setReducerClass(FlowCountSortReducer.class);//指定mapper輸出數(shù)據的kv類型job.setMapOutputKeyClass(FlowBean.class);job.setMapOutputValueClass(Text.class);//指定最終輸出的數(shù)據的kv類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//指定job的輸入原始文件所在目錄FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的輸出結果所在目錄 Path outPath = new Path(args[1]);/*FileSystem fs = FileSystem.get(conf);if(fs.exists(outPath)){fs.delete(outPath, true);}*/FileOutputFormat.setOutputPath(job, outPath);//將job中配置的相關參數(shù)，以及job所用的java類所在的jar包，提交給yarn去運行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);}}?
?
?
3.1.4 詳細流程示意圖? ? ? ? --------------這個圖非常重要
?------------------------Yarn拿到程序怎樣啟動，這是今天關心的問題---------------------------------
PS:
轉載于:https://www.cnblogs.com/bee-home/p/7908169.html
總結
以上是生活随笔為你收集整理的day08 MapReduce的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： requestURI的组成部分
- 下一篇： Netty之粘包分包