开发笔记:HadoopMapreduce

id

pid

amount

1001

01

1

1002

02

2

1003

03

3

id

pname

01

小米

02

华为

03

格力

id

pname

amount

1001

小米

1

1001

小米

1

1002

华为

2

1002

华为

2

1003

格力

3

1003

格力

3

package com.gec.mapreduce.table;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.Writable;

public class TableBean implements Writable {

       private String order_id; // 订单id

       private String p_id; // 产品id

       private int amount; // 产品数量

       private String pname; // 产品名称

       private String flag;// 表的标记

       public TableBean() {

              super();

       }

       public TableBean(String order_id, String p_id, int amount, String pname, String flag) {

              super();

              this.order_id = order_id;

              this.p_id = p_id;

              this.amount = amount;

              this.pname = pname;

              this.flag = flag;

       }

       public String getFlag() {

              return flag;

       }

       public void setFlag(String flag) {

              this.flag = flag;

       }

       public String getOrder_id() {

              return order_id;

       }

       public void setOrder_id(String order_id) {

              this.order_id = order_id;

       }

       public String getP_id() {

              return p_id;

       }

       public void setP_id(String p_id) {

              this.p_id = p_id;

       }

       public int getAmount() {

              return amount;

       }

       public void setAmount(int amount) {

              this.amount = amount;

       }

       public String getPname() {

              return pname;

       }

       public void setPname(String pname) {

              this.pname = pname;

       }

       @Override

       public void write(DataOutput out) throws IOException {

              out.writeUTF(order_id);

              out.writeUTF(p_id);

              out.writeInt(amount);

              out.writeUTF(pname);

              out.writeUTF(flag);

       }

       @Override

       public void readFields(DataInput in) throws IOException {

              this.order_id = in.readUTF();

              this.p_id = in.readUTF();

              this.amount = in.readInt();

              this.pname = in.readUTF();

              this.flag = in.readUTF();

       }

       @Override

       public String toString() {

              return order_id + "\\t" + p_id + "\\t" + amount + "\\t" ;

       }

}

package com.gec.mapreduce.table;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class TableMapper extends Mapper{

       TableBean bean = new TableBean();

       Text k = new Text();

       @Override

       protected void map(LongWritable key, Text value, Context context)

                     throws IOException, InterruptedException {

              // 1 获取输入文件类型

              FileSplit split = (FileSplit) context.getInputSplit();

              String name = split.getPath().getName();

              // 2 获取输入数据

              String line = value.toString();

              // 3 不同文件分别处理

              if (name.startsWith("order")) {// 订单表处理

                     // 3.1 切割

                     String[] fields = line.split(",");

                     // 3.2 封装bean对象

                     bean.setOrder_id(fields[0]);

                     bean.setP_id(fields[1]);

                     bean.setAmount(Integer.parseInt(fields[2]));

                     bean.setPname("");

                     bean.setFlag("0");

                     k.set(fields[1]);

              }else {// 产品表处理

                     // 3.3 切割

                     String[] fields = line.split(",");

                     // 3.4 封装bean对象

                     bean.setP_id(fields[0]);

                     bean.setPname(fields[1]);

                     bean.setFlag("1");

                     bean.setAmount(0);

                     bean.setOrder_id("");

                     k.set(fields[0]);

              }

              // 4 写出

              context.write(k, bean);

       }

}

package com.gec.mapreduce.table;

import java.io.IOException;

import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class TableReducer extends Reducer {

       @Override

       protected void reduce(Text key, Iterable values, Context context)

                     throws IOException, InterruptedException {

              // 1准备存储订单的集合

              ArrayList orderBeans = new ArrayList<>();

              // 2 准备bean对象

              TableBean pdBean = new TableBean();

              for (TableBean bean : values) {

                     if ("0".equals(bean.getFlag())) {// 订单表

                            // 拷贝传递过来的每条订单数据到集合中

                            TableBean orderBean = new TableBean();

                            try {

                                   BeanUtils.copyProperties(orderBean, bean);

                            } catch (Exception e) {

                                   e.printStackTrace();

                            }

                            orderBeans.add(orderBean);

                     } else {// 产品表

                            try {

                                   // 拷贝传递过来的产品表到内存中

                                   BeanUtils.copyProperties(pdBean, bean);

                            } catch (Exception e) {

                                   e.printStackTrace();

                            }

                     }

              }

              // 3 表的拼接

              for(TableBean bean:orderBeans){

                     bean.setP_id(pdBean.getPname());

                     // 4 数据写出去

                     context.write(bean, NullWritable.get());

              }

       }

}

package com.gec.mapreduce.table;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

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;

public class TableDriver {

       public static void main(String[] args) throws Exception {

              // 1 获取配置信息，或者job对象实例

              Configuration cOnfiguration= new Configuration();

              Job job = Job.getInstance(configuration);

              // 2 指定本程序的jar包所在的本地路径

              job.setJarByClass(TableDriver.class);

              // 3 指定本业务job要使用的mapper/Reducer业务类

              job.setMapperClass(TableMapper.class);

              job.setReducerClass(TableReducer.class);

              // 4 指定mapper输出数据的kv类型

              job.setMapOutputKeyClass(Text.class);

              job.setMapOutputValueClass(TableBean.class);

              // 5 指定最终输出的数据的kv类型

              job.setOutputKeyClass(TableBean.class);

              job.setOutputValueClass(NullWritable.class);

              // 6 指定job的输入原始文件所在目录

              FileInputFormat.setInputPaths(job, new Path(args[0]));

              FileOutputFormat.setOutputPath(job, new Path(args[1]));

              // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行

              boolean result = job.waitForCompletion(true);

              System.exit(result ? 0 : 1);

       }

}

1001       小米       1    

1001       小米       1    

1002       华为       2    

1002       华为       2    

1003       格力       3    

1003       格力       3