MapReduceJoin案例、ETL、压缩简介

Join案例

MapReduce里的join跟SQL很类似&＃xff0c;本质上还是联表查询&＃xff0c;即我们最终需要的信息存储在两张表里&＃xff0c;两张表有一个相同的字段&＃xff0c;从而可以关联起来。MapReduce里的join动作可以再Reduce端做也可以在Map端做&＃xff0c;数据量大时则最好在Map阶段做。

先看需求&＃xff1a;有一个订单表和一个商品表&＃xff0c;订单表里只有商品id&＃xff0c;没有商品名称&＃xff0c;商品名称在商品表里&＃xff0c;现在需要将订单表里的商品id替换为商品名称显示出来。

 如果是在Reduce端做join操作&＃xff0c;最好定义一个TableBean&＃xff08;名称随意&＃xff0c;随便取&＃xff0c;只要实现Writable接口即可&＃xff09;&＃xff0c;Map阶段的输出的key就是商品id&＃xff08;通过商品id将两张表关联起来&＃xff09;&＃xff0c;输出的value就是这个TableBean。在map阶段也要区分好每次读取的内容来自哪个文件&＃xff0c;可以通过setup的方法来完成&＃xff0c;setup方法会针对每个分片执行一次&＃xff0c;因此可以获取源文件路径&＃xff1a;

FileSplit fileSplit &＃61; (FileSplit) context.getInputSplit();
String fileName &＃61; fileSplit.getPath().getName();

然后在map方法中&＃xff0c;根据fileName&＃xff08;可定义为类的私有变量&＃xff09;来判断读的是订单表还是商品表&＃xff0c;如果是订单表&＃xff0c;设置这个TableBean的订单id、商品id、数量信息&＃xff0c;并且把标志位记为订单&＃xff0c;如果是商品表&＃xff0c;设置这个TableBean的商品id、商品名称&＃xff0c;并且把标志位记为商品。然后相同key&＃xff08;即相同商品id&＃xff09;的TableBean会进入到同一个reduce&＃xff0c;此时只要将商品TableBean的商品名称设给订单TableBean即可&＃xff0c;直接输出。

这种在reduce阶段进行join的操作可能会给reduce带来很大的压力&＃xff0c;因此推荐在map阶段就进行join。因为商品表数据不会很多&＃xff0c;所以考虑把商品表缓存起来&＃xff0c;在job提交阶段&＃xff0c;设置缓存文件为商品表&＃xff1a;

job.addCacheFile(new URI("xxx"));

同时也要设置ReduceTask数量为0&＃xff1a;

job.setNumReduceTasks(0);

在setup方法中获取缓存文件并读取&＃xff1a;

URI[] cacheFiles &＃61; context.getCacheFiles();
FsDataInputStream fis &＃61; FileSystem.get(context.getConfiguration()).open(new Path(cacheFiles[0]))
BufferedReader reader &＃61; new BufferedReader(new InputStreamReader(fis, "UTF-8"));
String line;
while(StringUtils.isNotEmpty(line &＃61; reader.readLine())) {
String[] fields &＃61; line.split("\t");
map.put(fields[0], fields[1]);
}
IOUtils.closeStream(reader);

ETL

ETL&＃xff0c;即Extract Transform Load&＃xff0c;描述数据从抽取&＃xff08;Extract&＃xff09;、转换&＃xff08;Transform&＃xff09;、加载&＃xff08;Load&＃xff09;到目的端的过程&＃xff0c;这期间往往需要数据清洗&＃xff0c;即清理掉不需要的数据。数据清洗一般只需要允许map程序&＃xff0c;不需要允许reduce程序&＃xff0c;一般配合正则表达式使用。

数据压缩

在map阶段的输入、输出&＃xff08;即reduce阶段的输入&＃xff09;、reduce阶段的输出&＃xff0c;文件格式通常为压缩文件&＃xff0c;这样可以节省磁盘空间、减少IO&＃xff0c;但是也增加了CPU的开销&＃xff0c;这种利弊需要在具体项目中进行权衡。常见的压缩算法对比&＃xff1a;

 这些算法的压缩性能比较&＃xff1a;

可以看出&＃xff0c;snappy压缩/解压速度最快&＃xff0c;但是压缩率不高&＃xff0c;bzip2压缩率最高&＃xff0c;但是速度最慢。压缩算法的选择可以参考下图&＃xff1a;

要想在hadoop中开启压缩&＃xff0c;需要配置如下参数&＃xff1a;

例如&＃xff0c;在map端输出开启压缩&＃xff1a;

conf.setBoolean("mapreduce.map.output.compress", true);
conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

这个设置之后可能没效果&＃xff0c;因为这只是map的输出压缩格式&＃xff0c;reduce端会自动解压&＃xff0c;修改reduce输出端的压缩格式才会看到输出的是压缩文件&＃xff1a;

FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);

Snappy的压缩格式需要结合Linux8使用&＃xff0c;如果在windows环境会报错。