MapReduce基础案例12-酒店数据清洗
问题描述
现有一个酒店数据的样本文件sample.csv,其内容如下所示。
sample.csv
SEQ,酒店,国家,省份,城市,商圈,星级,业务部门,房间数,图片数,评分,评论数 aba_2066,马尔康嘉绒大酒店,中国,四川,阿坝,,四星级/高档,OTA,85,,4.143799782,108 aba_2069,阿坝马尔康县澜峰大酒店,中国,四川,阿坝,,,低星,115,,3.977930069,129 aba_2094,阿坝鑫鸿大酒店,中国,四川,阿坝,四姑娘山,二星及其他,低星,,,, aba_2096,九寨沟管理局荷叶迎宾馆,中国,四川,阿坝,九寨沟沟口,二星及其他,低星,49,,3.972340107,394 aba_2097,九寨沟风景名胜区管理局贵宾楼饭店,中国,四川,阿坝,九寨沟沟口,三星级/舒适,低星,50,,4.12789011,585 aba_2098,九寨沟九鑫山庄,中国,四川,阿坝,九寨沟沟口,四星级/高档,OTA,60,,4.04046011,161 aba_2102,九寨沟冈拉美朵酒店,中国,四川,阿坝,九寨沟沟口,四星级/高档,OTA,198,,3.471659899,12 aba_2109,若尔盖大藏酒店古格王朝店,中国,四川,阿坝,西部旅游牧场,五星级/豪华,OTA,94,,3.263220072,62 aba_2111,若尔盖大藏酒店圣地店,中国,四川,阿坝,西部旅游牧场,四星级/高档,OTA,188,,3.921580076,119 aba_2117,九寨沟保利新九寨宾馆,中国,四川,阿坝,漳扎镇,五星级/豪华,OTA,329,,4.353809834,269 aba_2134,九寨沟名人酒店,中国,四川,阿坝,漳扎镇,三星级/舒适,低星,292,,4.539999962,57 aba_2150,九寨沟仁智度假酒店,中国,四川,阿坝,九寨沟沟口,三星级/舒适,低星,137,,3.782749891,173 aba_2152,九寨沟药泉山庄,中国,四川,阿坝,黄龙机场、川主寺,四星级/高档,OTA,128,,3.821099997,310 aba_2156,松潘黄龙寺华龙山庄,中国,四川,阿坝,黄龙风景区,四星级/高档,OTA,154,,3.315500021,107 aba_2213,阿坝山之旅背包客栈,中国,四川,阿坝,四姑娘山,二星及其他,客栈,20,,,13 aba_2217,阿坝若尔盖大酒店,中国,四川,阿坝,,二星及其他,低星,71,,4.267769814,2 aba_2233,九寨沟云天海大酒店,中国,四川,阿坝,九寨沟沟口,三星级/舒适,低星,100,,2.783930063, aba_2243,阿坝九旅假日酒店,中国,四川,阿坝,九寨沟沟口,四星级/高档,OTA,140,,3.00515008,49 aba_2248,九寨沟川主寺岷江源大酒店,中国,四川,阿坝,黄龙机场、川主寺,,低星,228,,3.211859941,109
在进一步分析之前,我们需要对该数据集进行清洗。要求如下:
- 将采集错误的数据剔除;
- 将任意关键字段为空的条目剔除;
- 关键字段定义为{星级、评分、评论数}。
实现思路
根据要求,我们只需要在并发的map阶段进行清洗即可。因为没有数据汇总或去重、排序等其他要求,所以不需要reduce阶段。
一、创建Java Maven项目
Maven依赖:
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>HadoopDemo</groupId> <artifactId>com.xueai8</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <!--hadoop依赖--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>3.3.1</version> </dependency> <!--hdfs文件系统依赖--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>3.3.1</version> </dependency> <!--MapReduce相关的依赖--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>3.3.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-jobclient</artifactId> <version>3.3.1</version> </dependency> <!--junit依赖--> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <!--编译器插件用于编译拓扑--> <plugin> <groupId>org.apache.maven.plugins</groupId> <!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3--> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>1.8</source> <!-- 源代码使用的JDK版本 --> <target>1.8</target> <!-- 需要生成的目标class文件的编译版本 --> <encoding>UTF-8</encoding><!-- 字符集编码 --> </configuration> </plugin> </plugins> </build> </project>
HotelMapper.java:
package com.xueai8.hotel; import org.apache.commons.lang3.StringUtils; 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 org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; /** * * Mapper 类 */ public class HotelMapper extends Mapper<LongWritable,Text,Text,NullWritable> { // 日志记录器 private final static Logger LOGGER = LoggerFactory.getLogger(HotelMapper.class); // 设置可重用的Text类型 private final static Text keyText = new Text(); // 每读入一行文本(一行酒店记录),就会调用下面这个函数进行处理(清洗转换) // 向map传入的value参数,就是要处理的一行记录 @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); // 先将Hadoop的Text类型转换为Java的String类型 String[] arr = line.split(",", -1); // 对一条记录进行拆分 // 过滤标题行 if("SEQ".equals(arr[0])){ return; } // 过滤字段不全的数据 if(arr.length < 12){ LOGGER.warn("删除错误行:{}", line); return; } String star = arr[6]; // 星级字段 String score = arr[10]; // 评分字段 String comment = arr[11]; // 评论数字段 // 将字段{星级、评论数、评分}中任意字段为空的数据删除, if(StringUtils.isBlank(star) || StringUtils.isBlank(score) || StringUtils.isBlank(comment) ){ LOGGER.warn("关键字段缺失,删除1行:{}", line); return; } // 将整理过后的数据写出 // 写出时,要从Java的String类型转换回Hadoop的Text类型 keyText.set(StringUtils.join(arr, ",")); context.write(keyText, NullWritable.get()); } }
HotelDriver.java:
package com.xueai8.hotel; 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.Counter; import org.apache.hadoop.mapreduce.Counters; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; /** * * 数据清洗 */ public class HotelDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { if (args.length < 2) { System.err.println("语法: HotelDriver <in> <out>"); System.exit(1);// 非正常退出 } // 指定hdfs相关的参数 Configuration conf = new Configuration(); Job job = Job.getInstance(conf,"data clean"); // 设置jar包所在路径 job.setJarByClass(HotelDriver.class); // 设置Mapper类 job.setMapperClass(HotelMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); // 只有mapper,没有reducer job.setNumReduceTasks(0); // 指定输入输出的解析类(默认就是Text文本,所以可省略这两句) job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 最后提交任务 boolean waitForCompletion = job.waitForCompletion(true); System.exit(waitForCompletion ? 0 : 1); } }
二、配置log4j
在src/main/resources目录下新增log4j的配置文件log4j.properties,内容如下:
log4j.rootLogger = info,stdout ### 输出信息到控制抬 ### log4j.appender.stdout = org.apache.log4j.ConsoleAppender log4j.appender.stdout.Target = System.out log4j.appender.stdout.layout = org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
三、项目打包
打开IDEA下方的终端窗口terminal,执行"mvn clean package"打包命令,如下图所示:
如果一切正常,会提示打jar包成功。如下图所示:
这时查看项目结构,会看到多了一个target目录,打好的jar包就位于此目录下。如下图所示:
四、项目部署
请按以下步骤执行。
1、启动HDFS集群和YARN集群。在Linux终端窗口中,执行如下的脚本:
$ start-dfs.sh $ start-yarn.sh
查看进程是否启动,集群运行是否正常。在Linux终端窗口中,执行如下的命令:
$ jps
这时应该能看到有如下5个进程正在运行,说明集群运行正常:
5542 NodeManager 5191 SecondaryNameNode 4857 NameNode 5418 ResourceManager 4975 DataNode
2、将数据文件sample.txt上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put sample.txt /data/mr/ $ hdfs dfs -ls /data/mr/
3、提交作业到Hadoop集群上运行。(如果jar包在Windows下,请先拷贝到Linux中。)
在终端窗口中,执行如下的作业提交命令:
$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.hotel.HotelDriver /data/mr /data/mr-output
4、查看输出结果。
在终端窗口中,执行如下的HDFS命令,查看输出结果:
$ hdfs dfs -ls /data/mr-output
可以看到如下的输出结果:
aba_2066,马尔康嘉绒大酒店,中国,四川,阿坝,,四星级/高档,OTA,85,,4.143799782,108 aba_2096,九寨沟管理局荷叶迎宾馆,中国,四川,阿坝,九寨沟沟口,二星及其他,低星,49,,3.972340107,394 aba_2097,九寨沟风景名胜区管理局贵宾楼饭店,中国,四川,阿坝,九寨沟沟口,三星级/舒适,低星,50,,4.12789011,585 aba_2098,九寨沟九鑫山庄,中国,四川,阿坝,九寨沟沟口,四星级/高档,OTA,60,,4.04046011,161 aba_2109,若尔盖大藏酒店古格王朝店,中国,四川,阿坝,西部旅游牧场,五星级/豪华,OTA,94,,3.263220072,62 aba_2111,若尔盖大藏酒店圣地店,中国,四川,阿坝,西部旅游牧场,四星级/高档,OTA,188,,3.921580076,119 aba_2117,九寨沟保利新九寨宾馆,中国,四川,阿坝,漳扎镇,五星级/豪华,OTA,329,,4.353809834,269 aba_2134,九寨沟名人酒店,中国,四川,阿坝,漳扎镇,三星级/舒适,低星,292,,4.539999962,57 aba_2150,九寨沟仁智度假酒店,中国,四川,阿坝,九寨沟沟口,三星级/舒适,低星,137,,3.782749891,173 aba_2152,九寨沟药泉山庄,中国,四川,阿坝,黄龙机场、川主寺,四星级/高档,OTA,128,,3.821099997,310 aba_2156,松潘黄龙寺华龙山庄,中国,四川,阿坝,黄龙风景区,四星级/高档,OTA,154,,3.315500021,107 aba_2217,阿坝若尔盖大酒店,中国,四川,阿坝,,二星及其他,低星,71,,4.267769814,2 aba_2243,阿坝九旅假日酒店,中国,四川,阿坝,九寨沟沟口,四星级/高档,OTA,140,,3.00515008,49