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

《Spark原理深入与编程实战》