说明

统计各个充电桩的 充电量、充电次数、充电时长

准备数据

充电桩编号,充电量,开始时间,结束时间

1,100,2020-01-01 09:30:50,2020-01-01 09:40:50
2,200,2020-01-11 10:40:50,2020-01-11 11:05:50
3,150,2020-01-01 11:20:50,2020-01-01 11:32:50
1,300,2020-01-03 14:30:50,2020-01-03 15:20:10
1,50,2020-02-02 12:30:50,2020-02-02 12:35:50
2,500,2020-02-02 15:30:30,2020-02-02 17:20:30
3,150,2020-02-01 11:20:50,2020-02-01 11:32:50
3,150,2020-02-03 11:20:50,2020-02-03 11:32:50
1,100,2020-03-01 09:30:50,2020-03-01 09:40:50
1,100,2020-03-05 09:30:50,2020-03-05 09:40:50
2,500,2020-03-12 15:30:30,2020-03-12 17:30:30
2,500,2020-03-22 15:30:30,2020-03-22 17:30:30
3,400,2020-03-02 17:30:20,2020-03-02 18:20:20
4,100,2020-04-01 11:20:50,2020-04-01 11:25:50
4,50,2020-04-21 11:20:50,2020-04-21 11:28:50
5,150,2020-04-11 11:20:50,2020-04-11 11:32:50
6,150,2020-04-15 11:20:50,2020-04-15 11:40:50

在hdfs创建目录

访问 http://hadoop1:9870 ，创建 /chongdianzhuang 目录

上传文件

将上面数据上传到 hdfs 的 /chongdianzhuang 目录

创建maven工程

略

pom.xml

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>${hadoop-version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop-version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>${hadoop-version}</version>
</dependency>

java

Cdz类

该类封装数据，MapReduce执行后，将其序列化到文件

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

import org.apache.hadoop.io.Writable;

public class Cdz implements Writable {

    //充电桩编号
    private String id;
    //充电次数
    private int num;
    //用电量
    private int eleNum;
    //充电时长，单位：分钟
    private int time;

    //序列化
    @Override
    public void write(DataOutput out) throws IOException {
        //写的顺序与读的顺序相同，否则读写错乱
        out.writeUTF(id);
        out.writeInt(num);
        out.writeInt(eleNum);
        out.writeInt(time);
    }

    //反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        //写的顺序与读的顺序必须相同，否则读写错乱
        id=in.readUTF();
        num=in.readInt();
        eleNum=in.readInt();
        time=in.readInt();
    }

    //保存文件的格式
    @Override
    public String toString() {

        return id+","+num+","+eleNum+","+time;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public int getTime() {
        return time;
    }

    public void setTime(int time) {
        this.time = time;
    }

    public int getEleNum() {
        return eleNum;
    }

    public void setEleNum(int eleNum) {
        this.eleNum = eleNum;
    }

}

注意： write(DataOutput out) 和 readFields(DataInput in) 读写数据的顺序必须一致，否则数据错乱

mapper类

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

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper extends Mapper<LongWritable, Text, Text, Cdz> {

    private Text idText=new Text();
    private Cdz cdz=new Cdz();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Cdz>.Context context)
            throws IOException, InterruptedException {

        //读取一行
        String line=value.toString();
        //根据 , 拆分
        String[] array=line.split(",");

        //拆分数据
        String id=array[0];
        int eleNum=Integer.parseInt(array[1]);
        String startTime=array[2];
        String endTime=array[3];

        SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            //计算充电时长
            Date startDate = sdf.parse(startTime);
            Date endDate=sdf.parse(endTime);
            int time=(int)(endDate.getTime()-startDate.getTime())/1000/60;

            //封装
            cdz.setId(id);
            cdz.setNum(1);
            cdz.setEleNum(eleNum);
            cdz.setTime(time);

            idText.set(id);

            //写数据
            context.write(idText, cdz);
        } catch (ParseException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }



    }
}

reducer

import java.io.IOException;

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

public class MyReducer extends Reducer<Text, Cdz, Text, Cdz>{
    private Cdz cdz=new Cdz();
    @Override
    protected void reduce(Text arg0, Iterable<Cdz> arg1, Reducer<Text, Cdz, Text, Cdz>.Context arg2)
            throws IOException, InterruptedException {
        int numTotal=0;//充电总次数
        int eleNumTotal=0;//充电总量
        int timeTotal=0;//充电总时长

        //遍历，累加
        for(Cdz item:arg1){
            numTotal ++;
            eleNumTotal+= item.getEleNum();
            timeTotal += item.getTime();
        }
        //封装
        cdz.setId(arg0.toString());
        cdz.setNum(numTotal);
        cdz.setEleNum(eleNumTotal);
        cdz.setTime(timeTotal);
        //写数据
        arg2.write(arg0, cdz);
    }

}

driver

package chongdianzhuang;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main2 extends Configured implements Tool{
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        //通过ToolRunner类 执行 覆盖的run()方法
        int res=ToolRunner.run(config, new Main2(), args);
        //退出客户端
        System.exit(res);
    }


    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(),"mywordcount2");
        //设置jar加载路径，不加此代码有时可能会报错
        job.setJarByClass(Main2.class);
        //设置map和reduce类
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        //设置map输出 key/value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Cdz.class);
        //设置最终输出 key/value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Cdz.class);
        //根据输入参数 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //提交
//        job.submit();
        boolean result = job.waitForCompletion(true);
        return result?1:0;
    }
}

本地测试

执行时传入参数：

D:\bigdata\chongdianzhuang\chongdianzhuang.txt D:\bigdata\chongdianzhuang\res2

yarn执行

打包

上传到 linux

将 jar包上传到 linux

执行

hadoop jar chongdianzhuang-0.0.1-SNAPSHOT.jar chongdianzhuang.Main2 /chongdianzhuang /cdz_res

原文出处：/show_1IX2B9zmE0SE.html