MapReduce处理csv

た入场券 2023-10-04 16:32 20阅读 0赞

##### **MapReduce处理csv** #####

csv是由逗号“，”来分割的文件，在编写Mapper类的时候需要以“，”分割成一个个的数据

查看一下csv数据  
![在这里插入图片描述][20210716124442269.png]

以上是为了测试做的数据，要处理的结果就是经过mapreduce再原封不动的出来，因为是测试，所以内容不做任何处理

**需求分析**

因为MapReduce的输入和输出都是k，v键值对的形式，所以考虑将输出v封装成一个对象，对象属性按照csv文件进行设置

**注意：因为封装为了对象使用MapReduce处理，就需要考虑序列化和反序列化，同时还需要考虑输入和输出的类型**

介绍MapReduce中序列化和反序列化

**MapReduce的序列化和反序列化**

\*\*序列化：\*\*序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁  
盘（持久化）和网络传输。

\*\*反序列化：\*\*反序列化就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换  
成内存中的对象。

**自定义 bean 对象实现序列化接口（Writable ）**

具体实现 bean 对象序列化步骤如下 7 步。

1.  必须实现 Writable 接口
2.  反序列化时，需要反射调用空参构造函数，所以必须有空参构造
3.  重写序列化方法
    
        @Override
            public void write(DataOutput dataOutput) throws IOException {
              
        
                dataOutput.writeUTF(id);
                dataOutput.writeUTF(name);
                dataOutput.writeUTF(age);
                dataOutput.writeUTF(sex);
            }
4.  重写反序列化方法
    
        @Override
            public void readFields(DataInput dataInput) throws IOException {
              
        
                this.id = dataInput.readUTF();
                this.name = dataInput.readUTF();
                this.age = dataInput.readUTF();
                this.sex = dataInput.readUTF();
            }
5.  **注意反序列化的顺序和序列化的顺序完全一致**
6.  要想把结果显示在文件中，需要重写 toString()，可用"\\t"，","等分开，方便后续用。
    
        @Override
            public String toString() {
              
                return id + "," + name + "," + age + "," + sex;
            }
7.  如果需要将自定义的 bean 放在 key 中传输，则还需要实现 Comparable 接口，因为MapReduce 框中的 Shuffle 过程要求对 key 必须能排序。

**进行业务编写**

**封装对象CsvBean**

**在序列化和反序列化是，要根据属性类型选择序列化的类型，比如属性是String，就要选择writeUTF方法，反序列化时就是readUTF方法**

package com.gis507.test.CsvSplit;
    
    import org.apache.hadoop.io.Writable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class CsvBean implements Writable{
        
    
        private String id;
        private String name;
        private String age;
        private String sex;
    
    
        public CsvBean() {
        
        }
    
        public CsvBean(String id, String name, String age, String sex) {
        
            this.id = id;
            this.name = name;
            this.age = age;
            this.sex = sex;
        }
    
        public String getId() {
        
            return id;
        }
    
        public void setId(String id) {
        
            this.id = id;
        }
    
        public String getName() {
        
            return name;
        }
    
        public void setName(String name) {
        
            this.name = name;
        }
    
        public String getAge() {
        
            return age;
        }
    
        public void setAge(String age) {
        
            this.age = age;
        }
    
        public String getSex() {
        
            return sex;
        }
    
        public void setSex(String sex) {
        
            this.sex = sex;
        }
    
        @Override
        public String toString() {
        
            return id + "," + name + "," + age + "," + sex;
        }
    
        @Override
        public void write(DataOutput dataOutput) throws IOException {
        
    
            dataOutput.writeUTF(id);
            dataOutput.writeUTF(name);
            dataOutput.writeUTF(age);
            dataOutput.writeUTF(sex);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
        
    
            this.id = dataInput.readUTF();
            this.name = dataInput.readUTF();
            this.age = dataInput.readUTF();
            this.sex = dataInput.readUTF();
        }
    }

**Mapper类：CsvSplitMapper**

package com.gis507.test.CsvSplit;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.checkerframework.checker.units.qual.C;
    
    import java.io.IOException;
    
    public class CsvSplitMapper extends Mapper<LongWritable, Text,Text,CsvBean> {
        
    
        private Text outK = new Text();
        private CsvBean outV = new CsvBean();
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
    
            // 1获取一行转为String
            String line = value.toString();
    
            //2 按照逗号分割
            // 1,jeryy,18,male,1
            String[] csvComments = line.split(",");
    
            //3 获取需要的值
            String id = csvComments[0];
            String name = csvComments[1];
            String age = csvComments[2];
            String sex = csvComments[3];
    
            //4 封装到对象
            outV.setId(id);
            outV.setName(name);
            outV.setAge(age);
            outV.setSex(sex);
    
            outK.set(id);
    
            //5 写出
            context.write(outK,outV);
    
        }
    }

**Reducer类：CsvSplitReducer**

package com.gis507.test.CsvSplit;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class CsvSplitReducer extends Reducer<Text,CsvBean, Text,CsvBean> {
        
    
        @Override
        protected void reduce(Text key, Iterable<CsvBean> values, Context context) throws IOException, InterruptedException {
        
    
            for (CsvBean value : values) {
        
    
                context.write(key,value);
            }
        }
    }

**Driver类：CsvSplitDriver**

package com.gis507.test.CsvSplit;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    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.io.IOException;
    
    public class CsvSplitDriver {
        
        public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        
    
            //1 获取job对象
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            //2 关联Driver类
            job.setJarByClass(CsvSplitDriver.class);
    
            //3 关联Mapper和Reducer类
            job.setMapperClass(CsvSplitMapper.class);
            job.setReducerClass(CsvSplitReducer.class);
    
            //4 设置Map的输入输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(CsvBean.class);
    
            //5	设置最终的输入输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(CsvBean.class);
    
            //6	设置输入输出路径
            FileInputFormat.setInputPaths(job,new Path("D:\\AAUser\\dic\\Files\\testFile\\test.csv"));
            FileOutputFormat.setOutputPath(job,new Path("D:\\AAUser\\dic\\Files\\testFile1"));
    
            //7 提交job
            boolean result = job.waitForCompletion(true);
            System.exit(result?0:1);
        }
    }

**输出结果：**  
![在这里插入图片描述][20210716124459137.png]

[20210716124442269.png]: https://img-blog.csdnimg.cn/20210716124442269.png
[20210716124459137.png]: https://img-blog.csdnimg.cn/20210716124459137.png