mapreduce的使用

mapreduce的使用

以下案例写之前需要导入jar包依赖：

  
      
              org.apache.hadoop
              hadoop-client
              2.6.0

1.单词计数案例：

package com.xyz;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author 小勇子start
 * @create 2021-10-11 16:35
 */
public class WordCount {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf=new Configuration();
        Job job= Job.getInstance(conf);
		//这里是对jar包的类，map的类和reduce的类三者的映射
        job.setJarByClass(WordCount.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
		//这里是对map输出的关键字和值对应类型的映射
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
		//这里是对reduce输出的关键字和值对应类型的映射
        job.setOutputKeyClass(Text.class);
        job.setOutputKeyClass(IntWritable.class);

        //数据输入路径(此处为本地测试)
        FileInputFormat.setInputPaths(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\mapreduceTest\\src\\data\\test.txt"));
        //数据输出路径(此处为本地测试)
        FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\mapreduceTest\\src\\data\\out1"));

        boolean flag=job.waitForCompletion(true);
        System.exit(flag?0:1);
    }

    static class Map extends Mapper{
        /*Text的包有很多，注意导的是：org.apache.hadoop.io.Text
        在map中前两个数据类型基本固定
        后两个代表着要输出的类型，如果有reduce，则以该类型发送给reduce
        */没有reduce则直接以该类型输出到目标
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] str=value.toString().split(",");
            for (String s:str) {
                context.write(new Text(s),new IntWritable(1));
            }
      }
    }

    static class Reduce extends Reducer{
        /*
        
        前两个是map发过来关键字和值的类型
        后两个是reduce输出关键字和值的类型
        
        */
        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
           int len=0;
            for (Object s:values) {
                len++;
            }
            context.write(key,new IntWritable(len));
        }
    }
}

2.数据清洗案例：

package com.xyz2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
/**
 * @author 小勇子start
 * @create 2021-10-11 17:23
 */
public class RepeatProcessing {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf=new Configuration();
        Job job= Job.getInstance(conf);
        
        job.setJarByClass(RepeatProcessing.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setMapOutputValueClass(NullWritable.class);
        job.setMapOutputKeyClass(Text.class);

        job.setOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
		//此处为本地测试
        FileInputFormat.setInputPaths(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\mapreduceTest\\src\\data\\test2.txt"));
        //此处为本地测试
        FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\mapreduceTest\\src\\data\\out2"));
        
        boolean flag=job.waitForCompletion(true);
        long repeat=job.getCounters().findCounter(Count.repeatCount).getValue();
        
        System.out.println("重复的行数为："+repeat);
        
        System.exit(flag?0:1);
    }

    static enum Count{//枚举计数
        repeatCount
    }

    static class Map extends Mapper{

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value,NullWritable.get());
        }
    }

    static class Reduce extends Reducer{
        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            Counter counter=context.getCounter(Count.repeatCount);
            int len=0;
            for (Object s:values) {
                len++;
            }
            long val= counter.getValue();
            val+=len-1;
            counter.setValue(val);
            context.write(key,NullWritable.get());
        }
    }
}

3.topN案例：

package com.xyz;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.*;

/**
 * @author 小勇子start
 * @create 2021-09-30 19:53
 */
public class Task2_4 {

    static class TaskMap extends Mapper {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] line = value.toString().split(";");
            String filmName = line[0];//电影名称
            String filmType = line[6];//电影类型
            String[] types = filmType.split("/|、|，");
            for (String s : types) {
                s=s.trim();
                if (s.length() > 2) {

                    for (int i = 2; i < s.length(); i += 2) {
                        String newStr = s.substring(i - 2, i);
                        context.write(new Text(newStr), new Text(filmName));
                    }
                } else
                    context.write(new Text(s), new Text(filmName));
            }

        }
    }

    static class TaskReduce extends Reducer {
       ArrayList arrayList=new ArrayList();
        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            //此处用set是因为Iterable values中存储的值中有重复的电影名称，需要取不同名字的个数
            HashSet set=new HashSet();
            for (Text s : values) {
                set.add(s);
            }
            Object[] data={set.size(),key.toString()};
            arrayList.add(data);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            Object[] data=arrayList.toArray();
            Arrays.parallelSort(data, new Comparator() {
                @Override
                public int compare(Object o1, Object o2) {
                    Object[] o_1=(Object[]) o1;
                    Object[] o_2=(Object[]) o2;
                    return (int)o_2[0]-(int)o_1[0];
                }
            });
            for (Object d:data) {
                Object[] d1=(Object[]) d;
                context.write(new Text(d1[1].toString()),new IntWritable((int)d1[0]));
           }
            /*取前3个
            for (int i=0;i<3;i++) {
            Object[] d1=(Object[]) data[i];
            context.write(new Text(d1[1].toString()),new IntWritable((int)d1[0]));
        	}
            */
            
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(Task2_4.class);
        job.setMapperClass(TaskMap.class);
        job.setReducerClass(TaskReduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.setInputPaths(job, new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\作业\\7月25日晚上任务\\数据\\2-1"));
        FileOutputFormat.setOutputPath(job, new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\作业\\7月25日晚上任务\\数据\\2-42"));

        boolean flag = job.waitForCompletion(true);

        System.exit(flag ? 0 : 1);

    }
}



4.join案例（自定义Writable）


package com.xyz;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * @author 小勇子start
 * @create 2021-10-12 10:32
 */
public class JoinDemo {
    static class MyDataWritable implements Writable {
        private String flag;//标记
        private String data;//一行数据

        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeUTF(flag);
            dataOutput.writeUTF(data);
        }
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            this.flag=dataInput.readUTF();
            this.data=dataInput.readUTF();
        }
        public String getFlag(){
            return flag;
        }
        public void setFlag(String flag){
           this.flag=flag;
        }
        public String getData() {
            return data;
        }
        public void setData(String data) {
            this.data = data;
        }
    }

    static class Map extends Mapper{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] str=value.toString().split(",");
            int id=Integer.valueOf(str[0]);
            String fileName=((FileSplit)context.getInputSplit()).getPath().getName();
            MyDataWritable myData=new MyDataWritable();
            myData.setData(value.toString());
            if(fileName.contains("customers"))
               myData.setFlag("kh");
            else
                myData.setFlag("order");
            context.write(new IntWritable(id),myData);
        }
    }

    static class Reduce extends Reducer{
        @Override
        protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException {
            String khData="";
            String oData="";
            for (MyDataWritable m:values) {
                if (m.getFlag().equals("kh"))
                    khData=m.getData();
                else
                    oData=m.getData();
            }
            String newData=khData+","+oData;
            context.write(new Text(newData),NullWritable.get());
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf=new Configuration();
        Job job= Job.getInstance(conf);

        job.setJarByClass(JoinDemo.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setMapOutputValueClass(MyDataWritable.class);
        job.setMapOutputKeyClass(IntWritable.class);

        job.setOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
        //此处为本地测试
        FileInputFormat.setInputPaths(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\joinDemo\\DemoOne\\join\\"));
        //此处为本地测试
        FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\小勇子\\Desktop\\大数据培训\\练习\\joinDemo\\DemoOne\\join\\out1"));
        boolean flag=job.waitForCompletion(true);
        System.exit(flag?0:1);
    }
}




5.时间日期格式化


import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author 小勇子start
 * @create 2021-10-11 17:53
 */
public class DateTimeFormat {
    public static void main(String[] args) throws ParseException {
        String dateStr="2021.09.05";
        SimpleDateFormat sdf=new SimpleDateFormat("yyyy.MM.dd");
        SimpleDateFormat sdf1=new SimpleDateFormat("yyyy-MM-dd");

        Date oldDate=sdf.parse(dateStr);
        String newDateStr=sdf1.format(oldDate);
        Date newDate=sdf1.parse(newDateStr);

        System.out.println(oldDate);//结果：Sun Sep 05 00:00:00 CST 2021
        System.out.println(newDateStr);//结果：2021-09-05
        System.out.println(newDate);//结果：Sun Sep 05 00:00:00 CST 2021
    }
}




6.mapreduce项目打jar包


    

  1. 右键项目 Open Module Settings
    2. 

  2. 找到左侧 Artifacts，点击中间的 +号，选择JAR,然后选择“”“From Module With dependicies”
    3. 

  3. 选择要执行的Main方法
    4. 

  4. 点击idea上面的Builde菜单，选择Builder Artifacts。然后builder即可（如果改了代码，直接点击Rebuild
    5. 

  5. 打包后将jar包上传至linux中
    6. 



7.运行jar包


    

  1. 在linux中创建一个文本文件，然后上传至hdfs中
    2. 

  2. 开始运行程序：  hadoop jar xxxx.jar com.xyzy.test1.MyWordDriver /xxx.txt /out1
    3. 





如果你程序中输入，输出路径是写死的（即不是用的args[0]这样的方式），那么/xxx.txt 和/out1就不需要了  ;在上传后jar包所在的位置执行上述命令，否则jar包前面使用绝对路径。



						  
					  
						
							


						

							
							
			大数据							
						



						

						
						


						

						
					
					

					
					

						
相关

						

				
						
			

																
								

										
Hadoop与大数据平台

									
									

								 

        

								

							
							

																
								

										
大数据数学基础--用python实现微积分 20212311孙佳瑜

									
									

								 

        

								

							
							

																
								

										
大数据学习之旅1——HDFS版本演化

									
									

								 

        

								

							
							

																
								

										
大数据学习之旅2——从零开始搭hadoop完全分布式集群

									
									

								 

        

								

							
							

																
								

										
大数据学习之旅1——HDFS版本演化

									
									

								 

        

								

							
							

																
								

										
【大数据-HBase】HBase根据rowKey模糊查询

									
									

								 

        

								

							
							

																
								

										
【大数据-HBase】HBase根据rowKey模糊查询

									
									

								 

        

								

							
							

																
								

										
大数据集快速上传到colab方法分享

									
									

								 

        

								

							
							

																
								

										
《Hadoop大数据技术开发实战》新书上线

									
									

								 

        

								

							
							

																
								

										
大数据第一次作业

									
									

								 

        

								

							
							

																
								

										
大数据第二次作业

									
									

								 

        

								

							
							

																
								

										
大数据，十有八九挂在装环境上（4）

									
									

								 

        

								

							
							
            
            
            
                                    
           
  			
						

					

					
				

				

														
					
					

					
									

					
									   
标签

										

										
    
										

									

										
										

									

					

					
				


			
		
		
		
			


						





	

	
一品网 冀ICP备14022925号-6