測試數(shù)據(jù)：

編寫代碼：

• mapper類

  /*** @author 17616*/ public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {@Overridepublic void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 首先獲取一行數(shù)據(jù)String line = value.toString ();// 將行內(nèi)的單詞進(jìn)行切分，使用一個數(shù)組進(jìn)行保存，切分?jǐn)?shù)據(jù)時根據(jù)源數(shù)據(jù)得知可以使用空格的方式切分。String[] arr = line.split (" ");for (String str : arr) {context.write (new Text (str), new LongWritable (1));}} }

• reducer類

  /*** @author 17616*/ public class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {@Overridepublic void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {// 定義變量記錄單詞出現(xiàn)的次數(shù)long sum = 0;for (LongWritable val : values) {// 記錄總次數(shù)sum += val.get ();}// 輸出數(shù)據(jù)，key就是單詞，value就是在map階段這個單詞出現(xiàn)的總次數(shù)context.write (key, new LongWritable (sum));} }

• Driver類

  /*** @author 17616* 官方案例，計算統(tǒng)計*/ public class WordCountDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {// 獲取當(dāng)前的默認(rèn)配置Configuration conf = new Configuration ();// 獲取代表當(dāng)前mr作業(yè)的job對象Job job = Job.getInstance (conf);// 指定一下當(dāng)前程序的入口類job.setJarByClass (WordCountDriver.class);//指定當(dāng)前Mapper、Reducer任務(wù)的類job.setMapperClass (WordCountMapper.class);job.setReducerClass (WordCountReducer.class);//設(shè)置Mapper的結(jié)果類型job.setMapOutputKeyClass (Text.class);job.setMapOutputValueClass (LongWritable.class);// 設(shè)置Reducer的結(jié)果類型job.setOutputKeyClass (Text.class);job.setOutputValueClass (LongWritable.class);//設(shè)置待分析的文件夾路徑（linux的路徑地址）FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/mapreduce"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/mapreduce"));if (!job.waitForCompletion (true)) {return;}} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

編寫代碼：

• mapper類

  /*** @author 17616*/ public class HeightMapper extends Mapper<LongWritable, Text, Text, LongWritable> {@Overridepublic void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//獲取一段數(shù)據(jù)String line = value.toString ();//獲取年份String year = line.substring (8, 12);//獲取溫度(強(qiáng)制轉(zhuǎn)換一下)int t = Integer.parseInt (line.substring (18, 22));context.write (new Text (year),new LongWritable (t));} }

• reducer類

  /*** @author 17616*/ public class HeightReducer extends Reducer<Text, LongWritable, Text, LongWritable> {@Overridepublic void reduce(Text year, Iterable<LongWritable> t, Context context) throws IOException, InterruptedException {long max = 0;for (LongWritable data : t) {if (max < data.get ()) {max = data.get ();}}context.write (year, new LongWritable (max));} }

• Driver類

  /*** @author 17616* -求最大值*/ public class HeightDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {// 獲取當(dāng)前的默認(rèn)配置Configuration conf = new Configuration ();// 獲取代表當(dāng)前mr作業(yè)的job對象Job job = Job.getInstance (conf);// 指定一下當(dāng)前程序的入口類job.setJarByClass (HeightDriver.class);//指定當(dāng)前Mapper、Reducer任務(wù)的類job.setMapperClass (HeightMapper.class);job.setReducerClass (HeightReducer.class);//設(shè)置Mapper的結(jié)果類型job.setMapOutputKeyClass (Text.class);job.setMapOutputValueClass (LongWritable.class);// 設(shè)置Reducer的結(jié)果類型job.setOutputKeyClass (Text.class);job.setOutputValueClass (LongWritable.class);//設(shè)置待分析的文件夾路徑（linux的路徑地址）FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/wendu/"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/wendu"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

編寫代碼：

• mapper類

  /*** @Author zhangyong* @Date 2020/4/7 19:53* @Version 1.0*/ public class DisMapper extends Mapper<LongWritable,Text,Text,NullWritable> {@Overridepublic void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {/*** 其中value只是一個變量，此處被當(dāng)做key進(jìn)行輸出*/context.write (value,NullWritable.get ());} }

• reducer類

  /*** @Author zhangyong* @Date 2020/4/7 21:21* @Version 1.0*/ public class DisReducer extends Reducer<Text, NullWritable, Text, NullWritable> {@Overridepublic void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {context.write (key, NullWritable.get ());}}

• Driver類

  /*** @Author zhangyong* @Date 2020/4/7 21:32* @Version 1.0* 數(shù)據(jù)去重*/ public class DisDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);//設(shè)置Drive類job.setJarByClass (DisReducer.class);//設(shè)置Mapper、Reduce類job.setMapperClass (DisMapper.class);job.setReducerClass (DisReducer.class);//Mapper的輸出job.setMapOutputKeyClass (Text.class);job.setMapOutputValueClass (NullWritable.class);//地址FileInputFormat.setInputPaths (job,new Path ("hdfs://anshun115:9000/distinct"));FileOutputFormat.setOutputPath (job,new Path ("hdfs://anshun115:9000/result/distinct"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

編寫代碼：

• Bean類

  /*** @Author zhangyong* @Date 2020/4/10 8:01* @Version 1.0*/ public class FlowBean implements Writable {private String phone;private String name;private String addr;private long flow;/*** 序列化** @param dataOutput* @throws IOException*/@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF (phone);dataOutput.writeUTF (name);dataOutput.writeUTF (addr);dataOutput.writeLong (flow);}/*** 反序列化** @param dataInput* @throws IOException*/@Overridepublic void readFields(DataInput dataInput) throws IOException {this.phone = dataInput.readUTF ();this.name = dataInput.readUTF ();this.addr = dataInput.readUTF ();this.flow = dataInput.readLong ();}public String getPhone() {return phone;}public void setPhone(String phone) {this.phone = phone;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getAddr() {return addr;}public void setAddr(String addr) {this.addr = addr;}public long getFlow() {return flow;}public void setFlow(long flow) {this.flow = flow;}@Overridepublic String toString() {return "FlowBean{" +"phone='" + phone + '\'' +", name='" + name + '\'' +", addr='" + addr + '\'' +", flow=" + flow +'}';} }

• mapper類

  /*** @Author zhangyong* @Date 2020/4/10 8:10* @Version 1.0*/ public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {@Overridepublic void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//獲取行String line = value.toString ();/*** [13901000123,zk,bj,343]* phone = 13901000123;* name = zk;* addr = bj;* flow = 343;*/String[] info = line.split (" ");FlowBean flowBean = new FlowBean ();flowBean.setPhone (info[0]);flowBean.setName (info[1]);flowBean.setAddr (info[2]);flowBean.setFlow (Integer.parseInt (info[3]));context.write (new Text (flowBean.getName ()), flowBean);} }

• reducer類

  /*** @Author zhangyong* @Date 2020/4/10 8:23* @Version 1.0*/ public class FlowReducer extends Reducer<Text, FlowBean, FlowBean, NullWritable> {@Overridepublic void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {FlowBean result = new FlowBean ();for (FlowBean value : values) {result.setPhone (value.getPhone ());result.setName (value.getName ());result.setAddr (value.getAddr ());result.setFlow (result.getFlow () + value.getFlow ());}context.write (result, NullWritable.get ());} }

• Driver類

  /*** @Author zhangyong* @Date 2020/4/10 8:28* @Version 1.0* 流量統(tǒng)計*/ public class FlowDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);// 加載主類job.setJarByClass (FlowDriver.class);//加載mapper、reduce類job.setMapperClass (FlowMapper.class);job.setReducerClass (FlowReducer.class);//設(shè)置map的的key、valuejob.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//設(shè)置輸出的的key、valuejob.setOutputKeyClass(FlowBean.class);job.setOutputValueClass (NullWritable.class);//設(shè)置路徑（傳輸、結(jié)果）FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/flow"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/flow"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

編寫代碼：

• Bean類

  /*** @Author zhangyong* @Date 2020/4/13 8:42* @Version 1.0*/ public class MovieBean implements WritableComparable<MovieBean> {private String name;private int hot;/*** 排序方法** @param o* @return*/@Overridepublic int compareTo(MovieBean o) {return o.hot - this.hot;}/*** 序列化** @param dataOutput* @throws IOException*/@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF (name);dataOutput.writeInt (hot);}/*** 反序列化** @param dataInput* @throws IOException*/@Overridepublic void readFields(DataInput dataInput) throws IOException {this.name = dataInput.readUTF ();this.hot = dataInput.readInt ();}public void setName(String name) {this.name = name;}public int getHot() {return hot;}public void setHot(int hot) {this.hot = hot;}@Overridepublic String toString() {return "MovieBean{" +"name='" + name + '\'' +", hot=" + hot +'}';} }

• mapper類

  /*** @Author zhangyong* @Date 2020/4/13 8:52* @Version 1.0*/ public class MovieMapper extends Mapper<LongWritable, Text, MovieBean, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//獲取一行String line = value.toString ();//截取數(shù)據(jù)String[] split = line.split (" ");//封裝對象MovieBean movieBean = new MovieBean ();movieBean.setName (split[0]);movieBean.setHot (Integer.parseInt (split[1]));//輸出context.write (movieBean, NullWritable.get ());} }

• reducer類

  無

• Driver類

  /*** @Author zhangyong* @Date 2020/4/13 9:19* @Version 1.0*/ public class MovieDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);job.setJarByClass (MovieDriver.class);job.setMapperClass (MovieMapper.class);//加載map輸出類型和value的輸出類型job.setMapOutputKeyClass (MovieBean.class);job.setMapOutputValueClass (NullWritable.class);FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/sort"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/sort"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

chinese.txt

english.txt

math.txt

編寫代碼：

• Bean類

  /*** @Author zhangyong* @Date 2020/4/10 10:00* @Version 1.0*/ public class ScoreBean implements Writable {private String name;private int chinese;private int math;private int english;@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF(name);dataOutput.writeInt(chinese);dataOutput.writeInt(math);dataOutput.writeInt(english);}@Overridepublic void readFields(DataInput dataInput) throws IOException {this.name = dataInput.readUTF();this.chinese = dataInput.readInt();this.math = dataInput.readInt();this.english = dataInput.readInt();}public String getName() {return name;}public void setName(String name) {this.name = name;}public int getChinese() {return chinese;}public void setChinese(int chinese) {this.chinese = chinese;}public int getMath() {return math;}public void setMath(int math) {this.math = math;}public int getEnglish() {return english;}public void setEnglish(int english) {this.english = english;}@Overridepublic String toString() {return "StuScore{" +"name='" + name + '\'' +", chinese=" + chinese +", math=" + math +", english=" + english +'}';} }

• mapper類

  /*** @Author zhangyong* @Date 2020/4/10 10:03* @Version 1.0*/ public class ScoreMapper extends Mapper<LongWritable, Text, Text, ScoreBean> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 獲取一行數(shù)據(jù)String line = value.toString();String[] data = line.split(" ");ScoreBean ss = new ScoreBean();ss.setName(data[1]);/*** 注意，此處導(dǎo)包的時候不要導(dǎo)錯，應(yīng)該導(dǎo)入的是org.apache.hadoop.mapreduce.lib.input.FileSplit;* 通過獲取當(dāng)前map階段的MapTask處理的切片信息來獲取文件名。*/FileSplit split = (FileSplit) context.getInputSplit();if (split.getPath().getName().equals("chinese.txt")) {ss.setChinese(Integer.parseInt(data[2]));} else if (split.getPath().getName().equals("math.txt")) {ss.setMath(Integer.parseInt(data[2]));} else if (split.getPath().getName().equals("english.txt")) {ss.setEnglish(Integer.parseInt(data[2]));}context.write(new Text(ss.getName()), ss);} }

• reducer類

  /*** @Author zhangyong* @Date 2020/4/10 10:05* @Version 1.0*/ public class ScoreReducer extends Reducer<Text, ScoreBean, Text, ScoreBean> {@Overrideprotected void reduce(Text key, Iterable<ScoreBean> values, Context context) throws IOException, InterruptedException {ScoreBean resultScore = new ScoreBean();// 此處key.toSting中只有name一個值，因為在map階段的輸出key只有nameresultScore.setName(key.toString());for (ScoreBean value : values) {// result.setFlow(result.getFlow() + value.getFlow());// 語文成績分?jǐn)?shù)resultScore.setChinese(resultScore.getChinese() + value.getChinese());// 數(shù)學(xué)成績分resultScore.setMath(resultScore.getMath() + value.getMath());// 英語成績分resultScore.setEnglish(resultScore.getEnglish() + value.getEnglish());}context.write(key, resultScore);} }

• Driver類

  /*** @Author zhangyong* @Date 2020/4/10 10:10* @Version 1.0* 統(tǒng)計成績*/ public class ScoreDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);//加載Drive類job.setJarByClass (ScoreDriver.class);//加載Mapper、Reducer類job.setMapperClass (ScoreMapper.class);job.setReducerClass (ScoreReducer.class);//加載map輸出類型和value的輸出類型job.setMapOutputKeyClass (Text.class);job.setMapOutputValueClass (ScoreBean.class);job.setOutputKeyClass (Text.class);job.setOutputValueClass (ScoreBean.class);FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/score"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/score"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

測試數(shù)據(jù)：

編寫代碼：

• Bean類

  /*** @Author zhangyong* @Date 2020/4/14 15:06* @Version 1.0*/ public class CountBean implements WritableComparable<CountBean> {private String name;private int count;@Overridepublic int compareTo(CountBean o) {return o.count - this.count;}/*** 序列化** @param dataOutput* @throws IOException*/@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF (name);dataOutput.writeInt (count);}/*** 反序列化** @param dataInput* @throws IOException*/@Overridepublic void readFields(DataInput dataInput) throws IOException {this.name = dataInput.readUTF ();this.count = dataInput.readInt ();}public String getName() {return name;}public void setName(String name) {this.name = name;}public int getCount() {return count;}public void setCount(int count) {this.count = count;}@Overridepublic String toString() {return "CountBean{" +"name='" + name + '\'' +", count=" + count +'}';} }

• mapper類

  OneCountMapper類

  /*** @Author zhangyong* @Date 2020/4/14 8:54* @Version 1.0*/ public class OneCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString ();/*** 通過 | 進(jìn)行切分時會得到一個數(shù)組，數(shù)組中的0號下標(biāo)為序號，1號下標(biāo)中有name和profit的數(shù)據(jù)。* 再通過切分1號下標(biāo)中的數(shù)據(jù)時可以獲取到name和profit的數(shù)據(jù)。*/String name = line.split ("\\|")[1].split (" ")[0];int count = Integer.parseInt (line.split ("\\|")[1].split (" ")[1]);context.write (new Text (name),new IntWritable (count));} }

  TwoCountMapper類

  /*** @Author zhangyong* @Date 2020/4/14 15:27* @Version 1.0*/ public class TwoCountMapper extends Mapper<LongWritable, Text, CountBean, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString ();/*** 因為第二個Mapper任務(wù)要讀取的數(shù)據(jù)內(nèi)容是第一個MR任務(wù)的結(jié)果文件，通常MR任務(wù)的結(jié)果文件是以TAB的方式來展示數(shù)據(jù)的。* 所以當(dāng)?shù)诙€Mapper任務(wù)要執(zhí)行切分時，所使用的分隔符應(yīng)該是\t——制表符。*/String name = line.split ("\t")[0];int count = Integer.parseInt (line.split ("\t")[1]);CountBean bean = new CountBean ();bean.setName (name);bean.setCount (count);context.write (bean, NullWritable.get ());} }

• reducer類

  /*** @Author zhangyong* @Date 2020/4/14 8:54* @Version 1.0*/ public class OneCountReducer extends Reducer<Text , IntWritable, Text, IntWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {//統(tǒng)計利潤int sum = 0;for (IntWritable value : values) {sum+=value.get ();}context.write (key,new IntWritable (sum));} }

• Driver類

  /*** @Author zhangyong* @Date 2020/4/14 8:54* @Version 1.0* job鏈操作數(shù)據(jù)*/ public class CountDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);job.setJarByClass (CountDriver.class);job.setMapperClass (OneCountMapper.class);job.setReducerClass (OneCountReducer.class);job.setMapOutputKeyClass (Text.class);job.setMapOutputValueClass (IntWritable.class);job.setOutputKeyClass (Text.class);job.setOutputValueClass (IntWritable.class);FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/count"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/count"));if (job.waitForCompletion (true)) {// 設(shè)置第二個Job任務(wù)Job job2 = Job.getInstance (conf);// 設(shè)置第二個Job任務(wù)的Mapperjob2.setMapperClass (TwoCountMapper.class);job2.setMapOutputKeyClass (CountBean.class);job2.setMapOutputValueClass (NullWritable.class);/*** 設(shè)置第二個Job任務(wù)是輸入輸出路徑。* 此處的輸入路徑是第一個job任務(wù)的輸出路徑* 注意設(shè)置路徑時，里面?zhèn)魅氲膉ob應(yīng)該是當(dāng)前的job任務(wù)，如下所示，應(yīng)該是job2。* 如果寫成前面的job任務(wù)名稱，在運(yùn)行時則會爆出錯誤，提示路徑不存在。*/FileInputFormat.setInputPaths (job2, new Path ("hdfs://anshun115:9000/result/count"));FileOutputFormat.setOutputPath (job2, new Path ("hdfs://anshun115:9000/result/count2"));// 此處提交任務(wù)時，注意用的是job2。job2.waitForCompletion (true);}}}

運(yùn)行結(jié)果：

count.txt

count2.txt

運(yùn)行結(jié)果：

part-r-00000

part-r-00001

part-r-00002

測試數(shù)據(jù)：

編寫代碼：

• Partitioner類

  /*** @Author zhangyong* @Date 2020/4/14 9:39* @Version 1.0* 全排序* 將上述文件內(nèi)容按照數(shù)字位數(shù)分別寫入三個文件，如下* 0-99的寫入到文件1* 100-999寫入到文件2* 1000-其他數(shù)據(jù)寫入到文件3*/ public class AutoPartitioner extends Partitioner<IntWritable, IntWritable> {@Overridepublic int getPartition(IntWritable key, IntWritable value, int numPartitions) {String num = String.valueOf (key.get ());if (num.matches ("[0-9][0-9]") || num.matches ("[0-9]")) {return 0;} else if (num.matches ("[0-9][0-9][0-9]")) {return 1;} else {return 2;}} }

• mapper類

  /*** @Author zhangyong* @Date 2020/4/14 9:44* @Version 1.0*/ public class NumSortMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString ();String[] data = line.split (" ");for (String num : data) {context.write (new IntWritable (Integer.parseInt (num)), new IntWritable (1));}} }

• reducer類

  /*** @Author zhangyong* @Date 2020/4/14 9:39* @Version 1.0*/ public class NumSortReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {@Overrideprotected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int result = 0;for (IntWritable count : values) {result = result + count.get ();}context.write (key, new IntWritable (result));} }

• Driver類

  /*** @Author zhangyong* @Date 2020/4/14 9:39* @Version 1.0**/ public class NumSortDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration ();Job job = Job.getInstance (conf);job.setJarByClass (NumSortDriver.class);job.setMapperClass (NumSortMapper.class);job.setMapOutputKeyClass (IntWritable.class);job.setMapOutputValueClass (IntWritable.class);/*** 由于結(jié)果文件系統(tǒng)是3個，所以需要在此指定Reduce的分區(qū)類和任務(wù)數(shù)。*/job.setPartitionerClass (AutoPartitioner.class);job.setNumReduceTasks (3);job.setReducerClass (NumSortReducer.class);job.setOutputKeyClass (IntWritable.class);job.setOutputValueClass (IntWritable.class);FileInputFormat.setInputPaths (job, new Path ("hdfs://anshun115:9000/numcount/"));FileOutputFormat.setOutputPath (job, new Path ("hdfs://anshun115:9000/result/numcount"));job.waitForCompletion (true);} }

運(yùn)行結(jié)果：

part-r-00000

part-r-00001

part-r-00002

測試數(shù)據(jù)：

編寫代碼：

• 第一個mapper類

  /*** @Author 張勇* @Site www.gz708090.com* @Version 1.0* @Date 2020-04-17 12:08*/ public class OneFriendMapper extends Mapper<LongWritable, Text, Text, Text> {/*** 輸入的key和value是根據(jù)文件內(nèi)容來確定。* 輸出的key和value是因為在業(yè)務(wù)邏輯中設(shè)定的輸出是name-friend好友關(guān)系。*/@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 獲取每行的數(shù)據(jù)String line = value.toString();// 獲取姓名String name = line.split(" ")[0];// 獲取好友String friend = line.split(" ")[1];context.write(new Text(name), new Text(friend));} }

• 第一個reducer類

  /*** @Author 張勇* @Site www.gz708090.com* @Version 1.0* @Date 2020-04-17 12:28*/ public class OneFriendReducer extends Reducer<Text, Text, Text, IntWritable> {/*** 輸入key和value要和mapper的輸出保持一致。* Text和IntWritable：* 如果是好友-1，如果不是好友就用-2。*/@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {ArrayList<String> friendList = new ArrayList<>();//處理好友關(guān)系for (Text value : values) {friendList.add(value.toString());if (key.toString().compareTo(value.toString()) < 0) {context.write(new Text(key + "-" + value), new IntWritable(1));} else {context.write(new Text(value + "-" + key), new IntWritable(1));}}// 處理可能相識的好友。for (int i = 0; i < friendList.size(); i++) {for (int j = 0; j < friendList.size(); j++) {String friend1 = friendList.get(i);String friend2 = friendList.get(j);if (friend1.compareTo(friend2) < 0) {context.write(new Text(friend1 + "-" + friend2), new IntWritable(2));}}}} }

• 第二個mapper類

  /*** @Author 張勇* @Site www.gz708090.com* @Version 1.0* @Date 2020-04-17 12:32*/ public class TwoFriendMapper extends Mapper<LongWritable, Text, Text, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 獲取一行數(shù)據(jù)String line = value.toString();// 獲取朋友關(guān)系的信息String friendInfo = line.split("\t")[0];// 獲取朋友關(guān)系的深度int deep = Integer.parseInt(line.split("\t")[1]);context.write(new Text(friendInfo), new IntWritable(deep));} }

• 第二個reducer類

  /*** @Author 張勇* @Site www.gz708090.com* @Version 1.0* @Date 2020-04-17 12:34*/ public class TwoFriendReducer extends Reducer<Text, IntWritable, Text, NullWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {Boolean flag = true;/*** 設(shè)定好友關(guān)系為true的時候進(jìn)行輸出* 因為題目要求是輸出可能相識的好友。所以為true的代碼應(yīng)該是2* 也就是好友關(guān)系為1的時候設(shè)置變量為false*/for (IntWritable value : values) {if (value.get() == 1) {flag = false;}}if (flag) {context.write(key, NullWritable.get());}} }

• Driver類

  /*** @Author 張勇* @Site www.gz708090.com* @Version 1.0* @Date 2020-04-17 12:36*/ public class FriendDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration();//設(shè)置第一輪MapReduce的相應(yīng)處理類與輸入輸出Job job1 = Job.getInstance(conf);job1.setJarByClass(FriendDriver.class);job1.setMapperClass(OneFriendMapper.class);job1.setReducerClass(OneFriendReducer.class);job1.setMapOutputKeyClass(Text.class);job1.setMapOutputValueClass(Text.class);job1.setOutputKeyClass(Text.class);job1.setOutputValueClass(IntWritable.class);//設(shè)置路徑（傳輸、結(jié)果）FileInputFormat.setInputPaths(job1, new Path("hdfs://anshun115:9000/friend"));FileOutputFormat.setOutputPath(job1, new Path("hdfs://anshun115:9000/result/friend"));//如果第一輪MapReduce完成再做這里的代碼if (job1.waitForCompletion(true)) {Job job2 = Job.getInstance(conf);// 設(shè)置第二個Job任務(wù)的Mapperjob2.setMapperClass(TwoFriendMapper.class);job2.setMapOutputKeyClass(Text.class);job2.setMapOutputValueClass(IntWritable.class);// 設(shè)置第二個Job任務(wù)的Reducerjob2.setReducerClass(TwoFriendReducer.class);job2.setOutputKeyClass(Text.class);job2.setOutputValueClass(NullWritable.class);/*** 設(shè)置第二個Job任務(wù)是輸入輸出路徑。* 此處的輸入路徑是第一個job任務(wù)的輸出路徑* 注意設(shè)置路徑時，里面?zhèn)魅氲膉ob應(yīng)該是當(dāng)前的job任務(wù)，如下所示，應(yīng)該是job2。* 如果寫成前面的job任務(wù)名稱，在運(yùn)行時則會爆出錯誤，提示路徑不存在。*/FileInputFormat.setInputPaths(job2, new Path("hdfs://anshun115:9000/result/friend"));FileOutputFormat.setOutputPath(job2, new Path("hdfs://anshun115:9000/result/friend2"));// 此處提交任務(wù)時，注意用的是job2。job2.waitForCompletion(true);}} }

運(yùn)行結(jié)果：

friend

friend2