一、实战概述

  • 本实战项目主要利用Hadoop MapReduce框架对多个文本文件中的IP地址进行整合并去除重复项。首先,在虚拟机上创建了三个包含IP地址列表的文本文件(ips01.txt、ips02.txt、ips03.txt),并将这些文件上传至HDFS上的/deduplicate/input目录作为原始数据。

  • 接着,通过IntelliJ IDEA创建了一个Maven项目MRDeduplicateIPs,并添加了hadoop-client和junit相关依赖。在项目中定义了三个关键类:DeduplicateIPsMapper、DeduplicateIPsReducer和DeduplicateIPsDriver。

  • DeduplicateIPsMapper类作为Map阶段的处理单元,读取每行输入文本数据(表示一个IP地址),将IP地址作为新的键输出,并使用NullWritable类型的空值,以准备后续去重操作。

  • DeduplicateIPsReducer类则负责Reduce阶段的逻辑,它接收Mapper阶段输出的所有具有相同IP地址的键值对,并通过不遍历值迭代器的方式实现键(即IP地址)的去重,确保每个唯一IP地址仅被写入一次。

  • 最后,DeduplicateIPsDriver类作为整个任务的驱动程序,负责配置和启动MapReduce作业。它设置了作业的输入与输出路径、Mapper和Reducer类,以及它们的键值类型。作业完成后,该类会遍历输出目录下的文件,读取并打印去重后的IP地址列表到控制台。

二、提出任务

  • 三个包含IP地址列表的文本文件(ips01.txt、ips02.txt、ips03.txt)
  • ips01.txt
192.168.1.1
172.16.0.1
10.0.0.1
192.168.1.2
192.168.1.3
172.16.0.2
10.0.0.2
192.168.1.1
172.16.0.1
10.0.0.3

  • ips02.txt
192.168.1.4
172.16.0.3
10.0.0.4
192.168.1.5
192.168.2.1
172.16.0.4
10.0.1.1
192.168.1.1
172.16.0.1 
10.0.0.1 

  • ips03.txt
192.168.1.6
172.16.1.1
10.0.2.1
192.168.1.7
192.168.3.1
172.16.0.5
10.0.0.5
192.168.1.1
172.16.0.1
10.0.0.3

  • 使用MR框架,实现网址去重
    在这里插入图片描述

三、完成任务

(一)准备数据

1、在虚拟机上创建文本文件

  • 在master虚拟机上使用文本编辑器创建三个文件:ips01.txt, ips02.txt, ips03.txt,并确保每个文件内存储的是纯文本格式的IP地址列表。
    在这里插入图片描述

2、上传文件到HDFS指定目录

  • 在master虚拟机上创建HDFS上的/deduplicate/input目录,用于存放待处理的原始数据文件。
  • 执行命令:hdfs dfs -mkdir -p /deduplicate/input
    在这里插入图片描述
  • 将本地创建的三个文本文件上传至HDFS的/deduplicate/input 目录
hdfs dfs -put ips01.txt /deduplicate/input/
hdfs dfs -put ips02.txt /deduplicate/input/
hdfs dfs -put ips03.txt /deduplicate/input/

  • 执行上述命令
    在这里插入图片描述

(二)实现步骤

  • 说明:集成开发环境IntelliJ IDEA版本 - 2022.3

1、创建Maven项目

  • Maven项目 - MRDeduplicateIPs,设置了JDK版本 - 1.8,组标识 - net.huawei.mr
    在这里插入图片描述
  • 单击【Create】按钮,得到初始化项目
    在这里插入图片描述

2、添加相关依赖

  • pom.xml文件里添加hadoop-clientjunit依赖
    在这里插入图片描述
<dependencies>                                   
    <!--hadoop客户端-->                             
    <dependency>                                 
        <groupId>org.apache.hadoop</groupId>     
        <artifactId>hadoop-client</artifactId>   
        <version>3.3.4</version>                 
    </dependency>                                
    <!--单元测试框架-->                                
    <dependency>                                 
        <groupId>junit</groupId>                 
        <artifactId>junit</artifactId>           
        <version>4.13.2</version>                
    </dependency>                                
</dependencies>                                                  

  • 刷新项目依赖
    在这里插入图片描述

3、创建日志属性文件

  • resources目录里创建log4j.properties文件
    在这里插入图片描述
log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/deduplicateips.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4、创建网址去重映射器类

  • 创建net.huawei.mr包,在包里创建DeduplicateIPsMapper
    在这里插入图片描述
package net.huawei.mr;

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;

/**
 * 功能:网址去重映射器类
 * 作者:华卫
 * 日期:2024年01月05日
 */
public class DeduplicateIPsMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        // 获取行内容
        String ip = value.toString();
        // 将<ip,null>键值对写入中间结果
        context.write(new Text(ip), NullWritable.get());
        
    }
}

  • 这段代码是Hadoop MapReduce编程框架中的一个Mapper类实现,名为DeduplicateIPsMapper,用于处理URL去重问题。虽然注释中提到的是“网址去重”,但实际代码逻辑仅针对IP地址进行操作。

  • 在Map阶段,该类继承自org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, NullWritable>

  • 输入键类型为LongWritable,通常表示文本行号;

  • 输入值类型为Text,存储一行原始数据(在这里应是IP地址);

  • 输出键类型为Text,用于输出去重后的IP地址;

  • 输出值类型为NullWritable,由于此处仅需去重并不需要具体值,所以使用空值。

  • map()方法是Mapper的主体逻辑部分,在每次调用时接收一行输入数据(键和值)。它首先将输入值(即每行文本内容)转换成字符串类型的IP地址,然后将这个IP地址作为新的键输出,并与NullWritable类型的空值一起写入到中间结果中。通过这种方式,Map阶段结束后,相同的IP地址会被归并到一起,以便后续Reducer阶段进一步处理以达到去重的目的。

5、创建网址去重归并器类

  • net.huawei.mr包里创建DeduplicateIPsReducer
    在这里插入图片描述
package net.huawei.mr;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * 功能:网址去重归并器类
 * 作者:华卫
 * 日期:2024年01月05日
 */
public class DeduplicateIPsReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) 
            throws IOException, InterruptedException {
        // 不遍历值迭代器,就可以实现键去重
        context.write(key, NullWritable.get());        
    }
}

  • 这段代码是Hadoop MapReduce编程框架中的一个Reducer类实现,名为DeduplicateIPsReducer,用于处理URL去重问题。尽管注释中提到的是“网址去重”,但实际代码逻辑只针对IP地址进行操作。

  • 在Reduce阶段,该类继承自org.apache.hadoop.mapreduce.Reducer<Text, NullWritable, Text, NullWritable>

  • 输入键类型为Text,存储Map阶段输出的去重后的IP地址;

  • 输入值类型为Iterable<NullWritable>,由于Mapper阶段输出的值为NullWritable,因此这里接收一组空值;

  • 输出键类型仍为Text,保持与Mapper阶段一致,输出去重后的唯一IP地址;

  • 输出值类型也仍为NullWritable,表示在这个任务中我们仅关注IP地址的去重,不需要额外信息。

  • reduce()方法是Reducer的核心逻辑部分,在此场景下,当多个相同的IP地址(键)被归并到一起时,无需遍历值迭代器(因为所有值都是NullWritable的空值),只需将接收到的每个唯一的IP地址作为键输出即可,从而达到去除重复IP的目的。通过这种方式,Reduce阶段结束后,输出结果中每个IP地址都只出现一次。

6、创建网址去重统计驱动器类

  • net.huawei.mr包里,创建DeduplicateIPsDriver
    在这里插入图片描述
package net.huawei.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
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.net.URI;

/**
 * 功能:网址去重驱动器类
 * 作者:华卫
 * 日期:2024年01月05日
 */
public class DeduplicateIPsDriver {
    public static void main(String[] args) throws Exception {
        // 创建配置对象
        Configuration conf = new Configuration();
        // 设置客户端使用数据节点主机名属性
        conf.set("dfs.client.use.datanode.hostname", "true");

        // 获取作业实例
        Job job = Job.getInstance(conf);
        // 设置作业启动类
        job.setJarByClass(DeduplicateIPsDriver.class);

        // 设置Mapper类
        job.setMapperClass(DeduplicateIPsMapper.class);
        // 设置map任务输出键类型
        job.setMapOutputKeyClass(Text.class);
        // 设置map任务输出值类型
        job.setMapOutputValueClass(NullWritable.class);

        // 设置Reducer类
        job.setReducerClass(DeduplicateIPsReducer.class);
        // 设置reduce任务输出键类型
        job.setOutputKeyClass(Text.class);
        // 设置reduce任务输出值类型
        job.setOutputValueClass(NullWritable.class);

        // 定义uri字符串
        String uri = "hdfs://master:9000";

        // 创建输入目录
        Path inputPath = new Path(uri + "/deduplicate/input");
        // 创建输出目录
        Path outputPath = new Path(uri + "/deduplicate/output");

        // 获取文件系统
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        // 删除输出目录(第二个参数设置是否递归)
        fs.delete(outputPath, true);

        // 给作业添加输入目录(允许多个)
        FileInputFormat.addInputPath(job, inputPath);
        // 给作业设置输出目录(只能一个)
        FileOutputFormat.setOutputPath(job, outputPath);

        // 等待作业完成
        job.waitForCompletion(true);

        // 输出统计结果
        System.out.println("======统计结果======");
        FileStatus[] fileStatuses = fs.listStatus(outputPath);
        for (int i = 1; i < fileStatuses.length; i++) {
            // 输出结果文件路径
            System.out.println(fileStatuses[i].getPath());
            // 获取文件系统数据字节输入流
            FSDataInputStream in = fs.open(fileStatuses[i].getPath());
            // 将结果文件显示在控制台
            IOUtils.copyBytes(in, System.out, 4096, false);
        }
    }
}

  • 这段代码是Hadoop MapReduce框架下的一个驱动器类(Driver)实现,名为DeduplicateIPsDriver,用于处理URL去重问题。它主要负责设置MapReduce作业的相关配置信息,并启动整个作业流程。
  1. 首先创建一个Hadoop Configuration对象并设置相关属性,如“dfs.client.use.datanode.hostname”,以便正确连接到HDFS数据节点。

  2. 初始化Job实例,并通过job.setJarByClass()方法指定作业的主类(即该驱动器类),使得Hadoop能够找到运行作业所需的JAR包。

  3. 设置作业的Mapper和Reducer类分别为DeduplicateIPsMapper和DeduplicateIPsReducer,同时设定它们的输入输出键值类型。

  4. 定义HDFS上输入与输出目录的URI路径,并使用FileSystem API获取文件系统实例,删除预先存在的输出目录以确保每次运行时结果都是新的。

  5. 将输入目录添加到作业中,设置唯一的输出目录。

  6. 调用job.waitForCompletion(true)方法启动并等待作业完成。

  7. 作业完成后,遍历输出目录下的所有文件(除成功标志文件外),打开每个文件并将其内容读取并打印到控制台,从而展示去重后的结果。

  • 总之,此驱动器类将配置、初始化及执行一个完整的MapReduce作业,该作业的主要功能是对存储在HDFS上的IP地址进行去重处理。

7、启动应用,查看结果

  • 运行DeduplicateIPsDriver类,查看结果
    在这里插入图片描述

四、实战总结

  • 本实战项目利用Hadoop MapReduce框架,通过自定义的DeduplicateIPsMapper和DeduplicateIPsReducer类处理三个文本文件中的IP地址数据。Mapper阶段读取每行IP并作为键输出,Reducer阶段对相同键(IP)进行归并去重。在DeduplicateIPsDriver驱动类中配置了作业属性、输入输出路径以及Map和Reduce阶段所使用的类,并成功执行了任务。最终,从原始文本数据中提取出一份不重复的IP地址集合。整个过程展示了MapReduce框架高效处理大规模数据集及实现特定业务逻辑的能力。
Logo

华为开发者空间,是为全球开发者打造的专属开发空间,汇聚了华为优质开发资源及工具,致力于让每一位开发者拥有一台云主机,基于华为根生态开发、创新。

更多推荐