一、實戰(zhàn)概述

• 本次實戰(zhàn)任務(wù)目標是使用Hadoop MapReduce技術(shù)對兩個包含重復(fù)數(shù)據(jù)的文本文件file1.txt和file2.txt進行去重操作，并將結(jié)果匯總到一個文件。首先啟動Hadoop服務(wù)，然后在虛擬機上創(chuàng)建這兩個文本文件并上傳到HDFS的/dedup/input目錄。

• 在Map階段，我們創(chuàng)建自定義Mapper類DeduplicateMapper，將TextInputFormat默認組件解析的鍵值對修改為需要去重的數(shù)據(jù)作為key，value設(shè)為空。在Reduce階段，我們創(chuàng)建自定義Reducer類DeduplicateReducer，直接復(fù)制輸入的key作為輸出的key，利用MapReduce默認機制對key（即文件中的每行內(nèi)容）進行自動去重。

• 我們還編寫MapReduce程序運行主類DeduplicateDriver，設(shè)置工作任務(wù)的相關(guān)參數(shù)，對HDFS上/dedup/input目錄下的源文件進行去重處理，并將結(jié)果輸出到HDFS的/dedup/output目錄。最后，運行DeduplicateDriver類，查看并下載結(jié)果文件，確認去重操作成功完成。此實戰(zhàn)任務(wù)展示如何運用Hadoop MapReduce進行大數(shù)據(jù)處理和去重操作，提升我們對分布式計算的理解和應(yīng)用能力。

二、提出任務(wù)

• 文件file1.txt本身包含重復(fù)數(shù)據(jù)，并且與file2.txt同樣出現(xiàn)重復(fù)數(shù)據(jù)，現(xiàn)要求使用Hadoop大數(shù)據(jù)相關(guān)技術(shù)對以上兩個文件進行去重操作，并最終將結(jié)果匯總到一個文件中。
  
  

• 編寫MapReduce程序，在Map階段采用Hadoop默認作業(yè)輸入方式后，將key設(shè)置為需要去重的數(shù)據(jù)，而輸出的value可以任意設(shè)置為空。

• 在Reduce階段，不需要考慮每一個key有多少個value，可以直接將輸入的key復(fù)制為輸出的key，而輸出的value可以任意設(shè)置為空，這樣就會使用MapReduce默認機制對key（也就是文件中的每行內(nèi)容）自動去重。

三、完成任務(wù)

（一）準備數(shù)據(jù)文件

• 啟動hadoop服務(wù)
  

1、在虛擬機上創(chuàng)建文本文件

• 創(chuàng)建兩個文本文件 - file1.txt、file2.txt
  

2、上傳文件到HDFS指定目錄

• 創(chuàng)建/dedup/input目錄，執(zhí)行命令：hdfs dfs -mkdir -p /dedup/input
  

• 將兩個文本文件 file1.txt與file2.txt，上傳到HDFS的/dedup/input目錄
  

（二）實現(xiàn)步驟

1、Map階段實現(xiàn)

• 使用IntelliJ開發(fā)工具創(chuàng)建Maven項目Deduplicate，并且新創(chuàng)建net.hw.mr包，在該路徑下編寫自定義Mapper類DeduplicateMapper，主要用于讀取數(shù)據(jù)集文件將TextInputFormat默認組件解析的類似<0，2022-11-1 a >鍵值對修改為<2022-11-1 a，null>。

（1）創(chuàng)建Maven項目

• Maven項目 - Deduplicate
  
• 單擊【Finish】按鈕
  

（2）添加相關(guān)依賴

• 在pom.xml文件里添加hadoop和junit依賴
  

<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）創(chuàng)建日志屬性文件

• 在resources目錄里創(chuàng)建log4j.properties文件
  

log4j.rootLogger=INFO, 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/deduplicate.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（4）創(chuàng)建去重映射器類

• 創(chuàng)建net.hw.mr包，在包里創(chuàng)建DeduplicateMapper類
  

package net.hw.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;/*** 功能：去重映射器類* 作者：華衛(wèi)* 日期：2022年11月30日*/
public class DeduplicateMapper extends Mapper<LongWritable, Text, Text, NullWritable> {private static Text field = new Text();// <0,2022-11-3 c> --> <2022-11-3 c,null>@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {field = value;context.write(field, NullWritable.get());}
}

2、Reduce階段實現(xiàn)

• 根據(jù)Map階段的輸出結(jié)果形式，同樣在net.hw.mr包下，自定義Reducer類DeduplicateReducer，主要用于接受Map階段傳遞來的數(shù)據(jù)，根據(jù)Shuffle工作原理，鍵值key相同的數(shù)據(jù)就會被合并，因此輸出數(shù)據(jù)就不會出現(xiàn)重復(fù)數(shù)據(jù)了。

創(chuàng)建去重歸并器類

• 在net.hw.mr包里創(chuàng)建DeduplicateReducer類
  

package net.hw.mr;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;/*** 功能：去重歸并器類* 作者：華衛(wèi)* 日期：2022年11月30日*/
public class DeduplicateReducer extends Reducer<Text, NullWritable, Text, NullWritable> {// <2022-11-3 c,null> <2022-11-4 d,null><2022-11-4 d,null>@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context)throws IOException, InterruptedException {context.write(key, NullWritable.get());}
}

3、Driver程序主類實現(xiàn)

• 編寫MapReduce程序運行主類DeduplicateDriver，主要用于設(shè)置MapReduce工作任務(wù)的相關(guān)參數(shù)，對HDFS上/dedup/input目錄下的源文件實現(xiàn)去重，并將結(jié)果輸入到HDFS的/dedup/output目錄下。

創(chuàng)建去重驅(qū)動器類

• 在net.hw.mr包里創(chuàng)建DeduplicateDriver類
  

package net.hw.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;/*** 功能：去重驅(qū)動器類* 作者：華衛(wèi)* 日期：2022年11月30日*/
public class DeduplicateDriver {public static void main(String[] args) throws Exception {// 創(chuàng)建配置對象Configuration conf = new Configuration();// 設(shè)置數(shù)據(jù)節(jié)點主機名屬性conf.set("dfs.client.use.datanode.hostname", "true");// 獲取作業(yè)實例Job job = Job.getInstance(conf);// 設(shè)置作業(yè)啟動類job.setJarByClass(DeduplicateDriver.class);// 設(shè)置Mapper類job.setMapperClass(DeduplicateMapper.class);// 設(shè)置map任務(wù)輸出鍵類型job.setMapOutputKeyClass(Text.class);// 設(shè)置map任務(wù)輸出值類型job.setMapOutputValueClass(NullWritable.class);// 設(shè)置Reducer類job.setReducerClass(DeduplicateReducer.class);// 設(shè)置reduce任務(wù)輸出鍵類型job.setOutputKeyClass(Text.class);// 設(shè)置reduce任務(wù)輸出值類型job.setOutputValueClass(NullWritable.class);// 定義uri字符串String uri = "hdfs://master:9000";// 創(chuàng)建輸入目錄Path inputPath = new Path(uri + "/dedup/input");// 創(chuàng)建輸出目錄Path outputPath = new Path(uri + "/dedup/output");// 獲取文件系統(tǒng)FileSystem fs = FileSystem.get(new URI(uri), conf);// 刪除輸出目錄fs.delete(outputPath, true);// 給作業(yè)添加輸入目錄FileInputFormat.addInputPath(job, inputPath);// 給作業(yè)設(shè)置輸出目錄FileOutputFormat.setOutputPath(job, outputPath);// 等待作業(yè)完成job.waitForCompletion(true);// 輸出統(tǒng)計結(jié)果System.out.println("======統(tǒng)計結(jié)果======");FileStatus[] fileStatuses = fs.listStatus(outputPath);for (int i = 1; i < fileStatuses.length; i++) {// 輸出結(jié)果文件路徑System.out.println(fileStatuses[i].getPath());// 獲取文件輸入流FSDataInputStream in = fs.open(fileStatuses[i].getPath());// 將結(jié)果文件顯示在控制臺IOUtils.copyBytes(in, System.out, 4096, false);}}
}

4、運行去重驅(qū)動器類，查看結(jié)果

• 運行DeduplicateDriver類
  
  
• 下載結(jié)果文件 - part-r-00000
  
• 查看結(jié)果文件 - part-r-00000
  

四、拓展練習(xí)

• 形式：單獨完成
• 題目：實現(xiàn)數(shù)據(jù)去重
• 要求：讓學(xué)生自己按照步驟實現(xiàn)數(shù)據(jù)去重的功能，以此來鞏固本節(jié)的學(xué)習(xí)內(nèi)容。寫一篇CSDN博客，記錄操作過程。

（一）原始問題

• 某人今天訪問很多不同的網(wǎng)站，移動或電信日志都會記錄在案，有些網(wǎng)站訪問次數(shù)多，有些網(wǎng)站訪問次數(shù)少，此人，今天訪問了多少個不同的網(wǎng)站？

（二）簡單化處理

• 假如有如下一些IP地址，分別保存在三個文件里，如何去掉重復(fù)地址？
• ips01.txt

192.168.234.21
192.168.234.22
192.168.234.21
192.168.234.21
192.168.234.23
192.168.234.21
192.168.234.21
192.168.234.21

• ips02.txt

192.168.234.25
192.168.234.21
192.168.234.21
192.168.234.26
192.168.234.21
192.168.234.27
192.168.234.21
192.168.234.27
192.168.234.21

• ips03.txt

192.168.234.29
192.168.234.21
192.168.234.26
192.168.234.21
192.168.234.25
192.168.234.25
192.168.234.21
192.168.234.22
192.168.234.21