文章目录
13.MapReduce框架原理13.7 数据清洗（ETL）13.7.1 需求13.7.1.1 输入数据13.7.1.2 期望输出数据13.7.2 需求分析13.7.3实现代码13.7.3.1 编写WebLogMapper类13.7.3.2编写WebLogDriver类13.7.3 运行结果13.8 MapReduce开发总结13.8.1 输入数据接口：InputFormat13.8.2 逻辑处理接口：Mapper13.8.3 Partitioner分区13.8.4 Comparable排序13.8.5 Combiner合并13.8.6 逻辑处理接口：Reducer13.8.7 输出数据接口：OutputFormat15. 常见错误及解决方案13.MapReduce框架原理
13.7 数据清洗（ETL）
“ETL，是英文Extract-Transform-Load的缩写，用来描述将数据从来源端经过抽取（Extract）、转换（Transform）、加载（Load）至目的端的过程。ETL一词较常用在数据仓库，但其对象并不限于数据仓库
在运行核心业务MapReduce程序之前，往往要先对数据进行清洗，清理掉不符合用户要求的数据。清理的过程往往只需要运行Mapper程序，不需要运行Reduce程序。
13.7.1 需求
去除日志中字段个数小于等于11的日志
13.7.1.1 输入数据
13.7.1.2 期望输出数据
每行字段长度都大于11
13.7.2 需求分析
需要在Map阶段对输入的数据根据规则进行过滤清洗。
13.7.3实现代码
创建一个weblog包
13.7.3.1 编写WebLogMapper类
创建一个WebLogMapper类
package com.summer.mapreduce.weblog;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/*** @author Redamancy* @create -10-08 10:06*/public class WebLogMapper extends Mapper<LongWritable, Text, Text, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {// 1 获取1行数据String line = value.toString();// 2 解析日志boolean result = parseLog(line, context);// 3 日志不合法退出if(!result){return;}// 4 日志合法就直接写出context.write(value, NullWritable.get());}// 2 封装解析日志的方法private boolean parseLog(String line, Mapper<LongWritable, Text, Text, NullWritable>.Context context) {// 1 截取String[] fields = line.split(" ");// 2 日志长度大于11的为合法if(fields.length > 11){return true;}else{return false;}}}
13.7.3.2编写WebLogDriver类
创建一个WebLogDriver类
package com.summer.mapreduce.weblog;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.NullWritable;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;import java.io.IOException;/*** @author Redamancy* @create -10-08 10:17*/public class WebLogDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// 输入输出路径需要根据自己电脑上实际的输入输出路径设置args = new String[] {"D:\\Acode\\Hadoop\\input\\inputweb","D:\\Acode\\Hadoop\\output\\outputweb"};// 1 获取job信息Configuration conf = new Configuration();Job job = Job.getInstance(conf);// 2 加载jar包job.setJarByClass(WebLogDriver.class);// 3 关联mapjob.setMapperClass(WebLogMapper.class);// 4 设置最终输出类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);// 设置reducetask个数为0job.setNumReduceTasks(0);// 5 设置输入和输出路径FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));// 6 提交boolean result = job.waitForCompletion(true);System.exit(result ? 0 : 1);}}
13.7.3 运行结果
处理前数据有14619行
处理后数据有13770行
说明代码没有错误
13.8 MapReduce开发总结
13.8.1 输入数据接口：InputFormat
（1）默认使用的实现类是：TextInputFormat
（2）TextInputFormat的功能逻辑是：一次读一行文本，然后将该行的起始偏移量作为key，行内容作为value返回。
（3）CombineTextInputFormat可以把多个小文件合并成一个切片处理，提高处理效率。
13.8.2 逻辑处理接口：Mapper
用户根据业务需求实现其中三个方法：map() setup() cleanup ()
13.8.3 Partitioner分区
（1）有默认实现 HashPartitioner，逻辑是根据key的哈希值和numReduces来返回一个分区号；key.hashCode()&Integer.MAXVALUE % numReduces
（2）如果业务上有特别的需求，可以自定义分区。
13.8.4 Comparable排序
（1）当我们用自定义的对象作为key来输出时，就必须要实现WritableComparable接口，重写其中的compareTo()方法。
（2）部分排序：对最终输出的每一个文件进行内部排序。
（3）全排序：对所有数据进行排序，通常只有一个Reduce。
（4）二次排序：排序的条件有两个。
13.8.5 Combiner合并
Combiner合并可以提高程序执行效率，减少IO传输。但是使用时必须不能影响原有的业务处理结果。
13.8.6 逻辑处理接口：Reducer
用户根据业务需求实现其中三个方法：reduce() setup() cleanup ()
13.8.7 输出数据接口：OutputFormat
（1）默认实现类是TextOutputFormat，功能逻辑是：将每一个KV对，向目标文本文件输出一行。
（2）用户还可以自定义OutputFormat。
15. 常见错误及解决方案
1）导包容易出错。尤其Text和CombineTextInputFormat。
2）Mapper中第一个输入的参数必须是LongWritable或者NullWritable，不可以是IntWritable. 报的错误是类型转换异常。
3）java.lang.Exception: java.io.IOException: Illegal partition for 13926435656 (4)，说明Partition和ReduceTask个数没对上，调整ReduceTask个数。
4）如果分区数不是1，但是reducetask为1，是否执行分区过程。答案是：不执行分区过程。因为在MapTask的源码中，执行分区的前提是先判断ReduceNum个数是否大于1。不大于1肯定不执行。
5）在Windows环境编译的jar包导入到Linux环境中运行，
hadoop jar wc.jar com.atguigu.mapreduce.wordcount.WordCountDriver /user/atguigu/ /user/atguigu/output
报如下错误：
Exception in thread “main” java.lang.UnsupportedClassVersionError: com/atguigu/mapreduce/wordcount/WordCountDriver : Unsupported major.minor version 52.0
原因是Windows环境用的jdk1.7，Linux环境用的jdk1.8。
解决方案：统一jdk版本。
6）缓存pd.txt小文件案例中，报找不到pd.txt文件
原因：大部分为路径书写错误。还有就是要检查pd.txt.txt的问题。还有个别电脑写相对路径找不到pd.txt，可以修改为绝对路径。
7）报类型转换异常。
通常都是在驱动函数中设置Map输出和最终输出时编写错误。
Map输出的key如果没有排序，也会报类型转换异常。
8）集群中运行wc.jar时出现了无法获得输入文件。
原因：WordCount案例的输入文件不能放用HDFS集群的根目录。
9）出现了如下相关异常
Exception in thread "main" java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Zat org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Native Method)at org.apache.hadoop.io.nativeio.NativeIO$Windows.access(NativeIO.java:609)at org.apache.hadoop.fs.FileUtil.canRead(FileUtil.java:977)java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:356)at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:371)at org.apache.hadoop.util.Shell.<clinit>(Shell.java:364)
解决方案：拷贝hadoop.dll文件到Windows目录C:\Windows\System32。个别同学电脑还需要修改Hadoop源码。
方案二：创建如下包名，并将NativeIO.java拷贝到该包名下
10）自定义Outputformat时，注意在RecordWirter中的close方法必须关闭流资源。否则输出的文件内容中数据为空。
@Overridepublic void close(TaskAttemptContext context) throws IOException, InterruptedException {if (atguigufos != null) {atguigufos.close();}if (otherfos != null) {otherfos.close();}}
