day08.MAPREDUCE详解【大数据教程】

单机版：内存受限，磁盘受限，运算能力受限分布式：1.文件分布式存储（HDFS）2.运算逻辑需要至少分成2个阶段（一个阶段独立并发，一个阶段汇聚）3.运算程序如何分发4.程序如何分配运算任务（切片）5.两阶段的程序如何启动？如何协调？6.整个程序运行过程中的监控？容错？重试？

1.MRAppMaster(mapreduce application master)2.MapTask3.ReduceTask

file1.txt 320Mfile2.txt 10M

file1.txt.split1-- 0~128file1.txt.split2-- 128~256file1.txt.split3-- 256~320file2.txt.split1-- 0~10M

minsize：默认值：1 配置参数： mapreduce.input.fileinputformat.split.minsize

//首先要定义四个泛型的类型//keyin: LongWritable valuein: Text//keyout: Text valueout:IntWritablepublic class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{//map方法的生命周期： 框架每传一行数据就被调用一次//key : 这一行的起始点在文件中的偏移量//value: 这一行的内容@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//拿到一行数据转换为stringString line = value.toString();//将这一行切分出各个单词String[] words = line.split(" ");//遍历数组，输出<单词，1>for(String word:words){context.write(new Text(word), new IntWritable(1));}}}

//生命周期：框架每传递进来一个kv 组，reduce方法被调用一次@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {//定义一个计数器int count = 0;//遍历这一组kv的所有v，累加到count中for(IntWritable value:values){count += value.get();}context.write(key, new IntWritable(count));}}

public class WordCountRunner {//把业务逻辑相关的信息（哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里……）描述成一个job对象//把这个描述好的job提交给集群去运行public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job wcjob = Job.getInstance(conf);//指定我这个job所在的jar包// wcjob.setJar("/home/hadoop/wordcount.jar");wcjob.setJarByClass(WordCountRunner.class);wcjob.setMapperClass(WordCountMapper.class);wcjob.setReducerClass(WordCountReducer.class);//设置我们的业务逻辑Mapper类的输出key和value的数据类型wcjob.setMapOutputKeyClass(Text.class);wcjob.setMapOutputValueClass(IntWritable.class);//设置我们的业务逻辑Reducer类的输出key和value的数据类型wcjob.setOutputKeyClass(Text.class);wcjob.setOutputValueClass(IntWritable.class);//指定要处理的数据所在的位置FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt");//指定处理完成之后的结果所保存的位置FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));//向yarn集群提交这个jobboolean res = wcjob.waitForCompletion(true);System.exit(res?0:1);}

public class TestSeri {public static void main(String[] args) throws Exception {//定义两个ByteArrayOutputStream，用来接收不同序列化机制的序列化结果ByteArrayOutputStream ba = new ByteArrayOutputStream();ByteArrayOutputStream ba2 = new ByteArrayOutputStream(); //定义两个DataOutputStream，用于将普通对象进行jdk标准序列化DataOutputStream dout = new DataOutputStream(ba);DataOutputStream dout2 = new DataOutputStream(ba2);ObjectOutputStream obout = new ObjectOutputStream(dout2);//定义两个bean，作为序列化的源对象ItemBeanSer itemBeanSer = new ItemBeanSer(1000L, 89.9f);ItemBean itemBean = new ItemBean(1000L, 89.9f); //用于比较String类型和Text类型的序列化差别Text atext = new Text("a");// atext.write(dout);itemBean.write(dout);byte[] byteArray = ba.toByteArray();//比较序列化结果System.out.println(byteArray.length);for (byte b : byteArray) {System.out.print(b);System.out.print(":");} System.out.println("-----------------------");String astr = "a";// dout2.writeUTF(astr);obout.writeObject(itemBeanSer);byte[] byteArray2 = ba2.toByteArray();System.out.println(byteArray2.length);for (byte b : byteArray2) {System.out.print(b);System.out.print(":");}}}

/** * 反序列化的方法，反序列化时，从流中读取到的各个字段的顺序应该与序列化时写出去的顺序保持一致 */@Overridepublic void readFields(DataInput in) throws IOException {upflow = in.readLong();dflow = in.readLong();sumflow = in.readLong();} /** * 序列化的方法 */@Overridepublic void write(DataOutput out) throws IOException out.writeLong(upflow);out.writeLong(dflow);//可以考虑不序列化总流量，因为总流量是可以通过上行流量和下行流量计算出来的out.writeLong(sumflow);}@Overridepublic int compareTo(FlowBean o) {//实现按照sumflow的大小倒序排序return sumflow>o.getSumflow()?-1:1;}

public class FlowBean implements WritableComparable<FlowBean>{long upflow;long downflow;long sumflow;//如果空参构造函数被覆盖，一定要显示定义一下，否则在反序列时会抛异常public FlowBean(){}public FlowBean(long upflow, long downflow) {super();this.upflow = upflow;this.downflow = downflow;this.sumflow = upflow + downflow;}public long getSumflow() {return sumflow;}public void setSumflow(long sumflow) {this.sumflow = sumflow;}public long getUpflow() {return upflow;}public void setUpflow(long upflow) {this.upflow = upflow;}public long getDownflow() {return downflow;}public void setDownflow(long downflow) {this.downflow = downflow;} //序列化，将对象的字段信息写入输出流@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upflow);out.writeLong(downflow);out.writeLong(sumflow);}//反序列化，从输入流中读取各个字段信息@Overridepublic void readFields(DataInput in) throws IOException {upflow = in.readLong();downflow = in.readLong();sumflow = in.readLong();}@Overridepublic String toString() {return upflow + "\t" + downflow + "\t" + sumflow;}@Overridepublic int compareTo(FlowBean o) {//自定义倒序比较规则return sumflow > o.getSumflow() ? -1:1;}}

public class FlowCount {static class FlowCountMapper extends Mapper<LongWritable, Text, FlowBean,Text > { @Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();String[] fields = line.split("\t");try {String phonenbr = fields[0];long upflow = Long.parseLong(fields[1]);long dflow = Long.parseLong(fields[2]);FlowBean flowBean = new FlowBean(upflow, dflow);context.write(flowBean,new Text(phonenbr));} catch (Exception e) {e.printStackTrace();}}} static class FlowCountReducer extends Reducer<FlowBean,Text,Text, FlowBean> { @Overrideprotected void reduce(FlowBean bean, Iterable<Text> phonenbr, Context context) throws IOException, InterruptedException {Text phoneNbr = phonenbr.iterator().next();context.write(phoneNbr, bean);}} public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(FlowCount.class);job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);job.setMapOutputKeyClass(FlowBean.class);job.setMapOutputValueClass(Text.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);// job.setInputFormatClass(TextInputFormat.class);FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}}

/** * 定义自己的从map到reduce之间的数据（分组）分发规则 按照手机号所属的省份来分发（分组）ProvincePartitioner * 默认的分组组件是HashPartitioner * @author */public class ProvincePartitioner extends Partitioner<Text, FlowBean> {static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();static {provinceMap.put("135", 0);provinceMap.put("136", 1);provinceMap.put("137", 2);provinceMap.put("138", 3);provinceMap.put("139", 4);} @Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {Integer code = provinceMap.get(key.toString().substring(0, 3));return code == null ? 5 : code;}}

Job job = Job.getInstance(conf);FileOutputFormat.setCompressOutput(job, true);FileOutputFormat.setOutputCompressorClass(job, (Class<? extends CompressionCodec>) Class.forName(""));

conf.setBoolean(Job.MAP_OUTPUT_COMPRESS, true);conf.setClass(Job.MAP_OUTPUT_COMPRESS_CODEC, GzipCodec.class, CompressionCodec.class);

public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException { FileSplit split = (FileSplit) genericSplit; Configuration job = context.getConfiguration(); this.maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE); start = split.getStart(); end = start + split.getLength(); final Path file = split.getPath(); // open the file and seek to the start of the split final FileSystem fs = file.getFileSystem(job); fileIn = fs.open(file); //根据文件后缀名创建相应压缩编码的codec CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file); if (null!=codec) { isCompressedInput = true; decompressor = CodecPool.getDecompressor(codec); //判断是否属于可切片压缩编码类型 if (codec instanceof SplittableCompressionCodec) { final SplitCompressionInputStream cIn = ((SplittableCompressionCodec)codec).createInputStream( fileIn, decompressor, start, end, SplittableCompressionCodec.READ_MODE.BYBLOCK); //如果是可切片压缩编码，则创建一个CompressedSplitLineReader读取压缩数据 in = new CompressedSplitLineReader(cIn, job, this.recordDelimiterBytes); start = cIn.getAdjustedStart(); end = cIn.getAdjustedEnd(); filePosition = cIn; } else {//如果是不可切片压缩编码，则创建一个SplitLineReader读取压缩数据，并将文件输入流转换成解压数据流传递给普通SplitLineReader读取 in = new SplitLineReader(codec.createInputStream(fileIn, decompressor), job, this.recordDelimiterBytes); filePosition = fileIn; } } else { fileIn.seek(start); //如果不是压缩文件，则创建普通SplitLineReader读取数据 in = new SplitLineReader(fileIn, job, this.recordDelimiterBytes); filePosition = fileIn; }

id

date

pid

amount

id

name

category_id

price

select a.id,a.date,b.name,b.category_id,b.price from t_order a join t_product b on a.pid = b.id

public class OrderJoin {static class OrderJoinMapper extends Mapper<LongWritable, Text, Text, OrderJoinBean> { @Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 拿到一行数据，并且要分辨出这行数据所属的文件String line = value.toString();String[] fields = line.split("\t");// 拿到itemidString itemid = fields[0];// 获取到这一行所在的文件名（通过inpusplit）String name = "你拿到的文件名";// 根据文件名，切分出各字段（如果是a，切分出两个字段，如果是b，切分出3个字段）OrderJoinBean bean = new OrderJoinBean();bean.set(null, null, null, null, null);context.write(new Text(itemid), bean);}}static class OrderJoinReducer extends Reducer<Text, OrderJoinBean, OrderJoinBean, NullWritable> { @Overrideprotected void reduce(Text key, Iterable<OrderJoinBean> beans, Context context) throws IOException, InterruptedException { //拿到的key是某一个itemid,比如1000//拿到的beans是来自于两类文件的bean// {1000,amount} {1000,amount} {1000,amount} --- {1000,price,name}//将来自于b文件的bean里面的字段，跟来自于a的所有bean进行字段拼接并输出}}}

public class TestDistributedCache {static class TestDistributedCacheMapper extends Mapper<LongWritable, Text, Text, Text>{FileReader in = null;BufferedReader reader = null;HashMap<String,String> b_tab = new HashMap<String, String>();String localpath =null;String uirpath = null;//是在map任务初始化的时候调用一次@Overrideprotected void setup(Context context) throws IOException, InterruptedException {//通过这几句代码可以获取到cache file的本地绝对路径，测试验证用Path[] files = context.getLocalCacheFiles();localpath = files[0].toString();URI[] cacheFiles = context.getCacheFiles();//缓存文件的用法——直接用本地IO来读取//这里读的数据是map task所在机器本地工作目录中的一个小文件in = new FileReader("b.txt");reader =new BufferedReader(in);String line =null;while(null!=(line=reader.readLine())){String[] fields = line.split(",");b_tab.put(fields[0],fields[1]);}IOUtils.closeStream(reader);IOUtils.closeStream(in);}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//这里读的是这个map task所负责的那一个切片数据（在hdfs上） String[] fields = value.toString().split("\t"); String a_itemid = fields[0]; String a_amount = fields[1]; String b_name = b_tab.get(a_itemid); // 输出结果 1001 98.9 banan context.write(new Text(a_itemid), new Text(a_amount + "\t" + ":" + localpath + "\t" +b_name ));}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(TestDistributedCache.class);job.setMapperClass(TestDistributedCacheMapper.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//这里是我们正常的需要处理的数据所在路径FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));//不需要reducerjob.setNumReduceTasks(0);//分发一个文件到task进程的工作目录job.addCacheFile(new URI("hdfs://hadoop-server01:9000/cachefile/b.txt"));//分发一个归档文件到task进程的工作目录// job.addArchiveToClassPath(archive);//分发jar包到task节点的classpath下// job.addFileToClassPath(jarfile); job.waitForCompletion(true);}}

public class WebLogBean {private String remote_addr;// 记录客户端的ip地址private String remote_user;// 记录客户端用户名称,忽略属性"-"private String time_local;// 记录访问时间与时区private String request;// 记录请求的url与http协议private String status;// 记录请求状态；成功是200private String body_bytes_sent;// 记录发送给客户端文件主体内容大小private String http_referer;// 用来记录从那个页面链接访问过来的private String http_user_agent;// 记录客户浏览器的相关信息private boolean valid = true;// 判断数据是否合法public String getRemote_addr() {return remote_addr;}public void setRemote_addr(String remote_addr) {this.remote_addr = remote_addr;}public String getRemote_user() {return remote_user;}public void setRemote_user(String remote_user) {this.remote_user = remote_user;}public String getTime_local() {return time_local;}public void setTime_local(String time_local) {this.time_local = time_local;}public String getRequest() {return request;}public void setRequest(String request) {this.request = request;}public String getStatus() {return status;}public void setStatus(String status) {this.status = status;}public String getBody_bytes_sent() {return body_bytes_sent;}public void setBody_bytes_sent(String body_bytes_sent) {this.body_bytes_sent = body_bytes_sent;}public String getHttp_referer() {return http_referer;}public void setHttp_referer(String http_referer) {this.http_referer = http_referer;}public String getHttp_user_agent() {return http_user_agent;}public void setHttp_user_agent(String http_user_agent) {this.http_user_agent = http_user_agent;}public boolean isValid() {return valid;}public void setValid(boolean valid) {this.valid = valid;}@Overridepublic String toString() { StringBuilder sb = new StringBuilder(); sb.append(this.valid); sb.append("\001").append(this.remote_addr); sb.append("\001").append(this.remote_user); sb.append("\001").append(this.time_local); sb.append("\001").append(this.request); sb.append("\001").append(this.status); sb.append("\001").append(this.body_bytes_sent); sb.append("\001").append(this.http_referer); sb.append("\001").append(this.http_user_agent); return sb.toString();}}

public class WebLogParser { public static WebLogBean parser(String line) { WebLogBean webLogBean = new WebLogBean(); String[] arr = line.split(" "); if (arr.length > 11) { webLogBean.setRemote_addr(arr[0]); webLogBean.setRemote_user(arr[1]); webLogBean.setTime_local(arr[3].substring(1)); webLogBean.setRequest(arr[6]); webLogBean.setStatus(arr[8]); webLogBean.setBody_bytes_sent(arr[9]); webLogBean.setHttp_referer(arr[10]); if (arr.length > 12) { webLogBean.setHttp_user_agent(arr[11] + " " + arr[12]); } else { webLogBean.setHttp_user_agent(arr[11]); } if (Integer.parseInt(webLogBean.getStatus()) >= 400) {// 大于400，HTTP错误 webLogBean.setValid(false); } } else { webLogBean.setValid(false); } return webLogBean; } public static String parserTime(String time) { time.replace("/", "-"); return time; }}

public class WeblogPreProcess { static class WeblogPreProcessMapper extends Mapper<LongWritable, Text, Text, NullWritable> {Text k = new Text();NullWritable v = NullWritable.get(); @Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString();WebLogBean webLogBean = WebLogParser.parser(line);if (!webLogBean.isValid())return;k.set(webLogBean.toString());context.write(k, v);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(WeblogPreProcess.class);job.setMapperClass(WeblogPreProcessMapper.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}}