reducejoin

public class TableBean implements Writable{
       //封装对应字段
       private String order_id;
       private String pid;
       private int amount;
       private String pname;
       private String flag;

pubic TableBean(){}

setter;
getter;

       public void write(DataOutput out) throws IOException{
           out.writeUTF(order_id);
           out.writeUTF(pid);
           out.writeInt(amount);
           out.writeUTF(pname);
           out.writeUTF(flag);
       }

       public void readFields(DataInput in) throws IOException{
           order_id = in.readUTF();
           pid = in.readUTF();
           amount = in.readInt();
           pname = in.readUTF();
           flag = in.readUTF();

       }
       public String toString(){
           return order_i+"\t+panme+"\t+amount
       }
   }

   public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean>{
       pubic void map(LongWritable key, Text value, Context context)
               throws IOException, InterruptException{
           TableBean v =new TableBean();
           Text k = new Text();

           //区分两张表
           FileSplit inputsplit = (FileSplit)context.getInputSplit();
           String name = inputsplit.getPath().getName();//得到全路径名

//获取数据
String line = value.toString();

           //区分此时是订单表
           if(name.contains("order.txt")){
               //切分字段
               String[] fields = line.split("\t");

               //封装对象
               v.setOrder_id(fields[0]);
               v.setPid(fields[1]);
               v.setAmount(Integer.parseInt(fields[2]));
               v.setPname("");
               v.setFlag("0");

//设置k 商品id
k.set(fields[1]);

           }else{//此时是商品表
               //切分字段
               String[] fields = line.split("\t");

               //封装对象
               v.setOrder_id("");
               v.setPid(fields[0]);
               v.setAmount(0);
               v.setPname(fields[1]);
               v.setFlag("1");

               //设置k
               k.set(fields[0]);
           }

context.write(k,v);

}

   pubic class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable>{
       public void reduce(Text key, Interable<Table> value, Context context)
               throws Exception{
           //创建集合存放订单数据
           ArrayList<TableBean> orderBean = new ArrayList<>();

//商品存储
TableBean pdBean = new TableBean();//把pd中商品名拷贝到orderBean

           for(TableBean v : value){
               if("0".equals(v.getFlag())){
                   //订单表
                   //1、创建一个临时变量拷贝数据
                   TableBean tableBean = new TableBean();

//2、拷贝
BeanUtils.copyProperties(tableBean, v);

orderBean.add(tableBean);

               }else{
                   //商品表
                   BeanUtils.copyProperties(pdBean, v);

               }
           }
           //拼接表
           for(TableBean tableBean : orderBean){
               //加入商品名
               tableBean.setpanem(pdBean.getPname());
               context.write(tableBean, NullWritable.get());
           }

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

job.setJarByClass(FileDriver.class);

job.setMapperClass(TableMapper.class);
job.setReducerClass(TableReducer.class);

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

job.setOutputKeyClass(TableBean.class);
job.setOutputValueClass(NullWritable.class);

FileInputFormat.setInputPaths(job, new Path("C:/in"));
FileOutputFormat.setOutputPath(job, new Path("c:/out"));

           job.waitForCompletion(true);
       }
   }

猜你喜欢