mapreduce程序案例2--order表和product表的join实现（pid是关联主键）

Bertha 。 2022-12-13 12:41 203阅读 0赞

order表和product表的join实现（pid是关联主键）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70][]

order.txt:

1001 20150710 P0001 2
    1002 20150710 P0001 3
    1003 20150710 P0002 3

product.txt:

P0001 小米5 C01 2
    P0002 锤子T1 C01 3

（1）reduce阶段的join代码：

package application.mapreduce;
    
    import lombok.AllArgsConstructor;
    import lombok.Builder;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.apache.hadoop.io.Writable;
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 7:38
     * @Version: 1.0
     * @Description:
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    public class OrderProduct implements Writable {
        private String date = "";
        private String pid = "";
        private int amount = 0;
        private String productName = "";
        private String categoryId = "";
        private double price = 0.0;
        private boolean flag = false;
    
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeUTF(date);
            dataOutput.writeUTF(pid);
            dataOutput.writeInt(amount);
            dataOutput.writeUTF(productName);
            dataOutput.writeUTF(categoryId);
            dataOutput.writeDouble(price);
            dataOutput.writeBoolean(flag);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            date = dataInput.readUTF();
            pid = dataInput.readUTF();
            amount = dataInput.readInt();
            productName = dataInput.readUTF();
            categoryId = dataInput.readUTF();
            price = dataInput.readDouble();
            flag = dataInput.readBoolean();
        }
    }

package application.mapreduce;
    
    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;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 7:46
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductMapper extends Mapper<LongWritable, Text,Text,OrderProduct> {
    
        private Text text = new Text();
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            FileSplit fileSplit = (FileSplit) context.getInputSplit();
            String fileName = fileSplit.getPath().getName();
            if ("order.txt".equalsIgnoreCase(fileName)){
                String[] fields = line.split(" ");
                String date = fields[1];
                String pid = fields[2];
                int amount = Integer.parseInt(fields[3]);
                OrderProduct orderProduct = OrderProduct.builder()
                        .date(date)
                        .pid(pid)
                        .amount(amount)
                        .productName("")
                        .categoryId("")
                        .price(0)
                        .flag(false)
                        .build();
                text.set(pid);
                context.write(text,orderProduct);
            }
            else{
                String[] fields = line.split(" ");
                String pid = fields[0];
                String productName = fields[1];
                String categoryId = fields[2];
                double price = Double.parseDouble(fields[3]);
                OrderProduct orderProduct = OrderProduct.builder()
                        .date("")
                        .pid(pid)
                        .amount(0)
                        .productName(productName)
                        .categoryId(categoryId)
                        .price(price)
                        .flag(true)
                        .build();
                text.set(pid);
                context.write(text,orderProduct);
            }
        }
    }

package application.mapreduce;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 8:05
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductReducer extends Reducer<Text,OrderProduct,OrderProduct, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<OrderProduct> values, Context context) throws IOException, InterruptedException {
            List<OrderProduct> orders = new ArrayList<>();
            List<OrderProduct> products = new ArrayList<>();
            List<OrderProduct> joinResult = new ArrayList<>();
    
            for (OrderProduct orderProduct : values){
                if (orderProduct.isFlag()){
                    products.add(orderProduct);
                }
                else{
                    orders.add(orderProduct);
                }
    
                System.out.println(orderProduct);
            }
    
            for (OrderProduct orderProduct : products){
                for (int i=0;i<orders.size();i++){
                    OrderProduct orderProduct2 = orders.get(i);
                    if (orderProduct.getPid().equalsIgnoreCase(orderProduct2.getPid())){
                        joinResult.add(OrderProduct.builder()
                                .date(orderProduct2.getDate())
                                .pid(orderProduct.getPid())
                                .amount(orderProduct2.getAmount())
                                .productName(orderProduct.getProductName())
                                .categoryId(orderProduct.getCategoryId())
                                .price(orderProduct.getPrice())
                                .build());
                    }
                }
            }
    
            joinResult.forEach(orderProduct -> {
                try {
                    context.write(orderProduct,NullWritable.get());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }

package application.mapreduce;
    
    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;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 8:19
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductDriver {
        public static void main(String[] args) throws Exception {
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            job.setJarByClass(OrderProductDriver.class);
    
            job.setMapperClass(OrderProductMapper.class);
            job.setReducerClass(OrderProductReducer.class);
    
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(OrderProduct.class);
    
            job.setOutputKeyClass(OrderProduct.class);
            job.setOutputValueClass(NullWritable.class);
    
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
    
            job.waitForCompletion(true);
        }
    }

（2）map阶段的join过程

package application.mapreduce;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.BufferedReader;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.URI;
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 7:46
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductMapper extends Mapper<LongWritable, Text,OrderProduct, NullWritable> {
    
        private Text text = new Text();
        private Map<String,OrderProduct> orderProductMap = new HashMap<>();
    
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            URI[] caches = context.getCacheFiles();
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(caches[0].getPath())));
            String line = null;
            while((line = bufferedReader.readLine())!=null){
                String[] fields = line.split(" ");
                String pid = fields[0];
                String productName = fields[1];
                String categoryId = fields[2];
                double price = Double.parseDouble(fields[3]);
                OrderProduct orderProduct = OrderProduct.builder()
                        .date("")
                        .pid(pid)
                        .amount(0)
                        .productName(productName)
                        .categoryId(categoryId)
                        .price(price)
                        .flag(true)
                        .build();
                orderProductMap.put(pid,orderProduct);
            }
    
            System.out.println(orderProductMap);
        }
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split(" ");
            String date = fields[1];
            String pid = fields[2];
            int amount = Integer.parseInt(fields[3]);
            OrderProduct orderProduct = OrderProduct.builder()
                    .date(date)
                    .pid(pid)
                    .amount(amount)
                    .productName("")
                    .categoryId("")
                    .price(0)
                    .flag(false)
                    .build();
            orderProductMap.forEach((key1, value1) -> {
                if (key1.equalsIgnoreCase(orderProduct.getPid())){
                    OrderProduct orderProduct1 = OrderProduct.builder()
                            .date(orderProduct.getDate())
                            .pid(value1.getPid())
                            .amount(orderProduct.getAmount())
                            .productName(value1.getProductName())
                            .categoryId(value1.getCategoryId())
                            .price(value1.getPrice())
                            .build();
                    try {
                        context.write(orderProduct1,NullWritable.get());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

package application.mapreduce;
    
    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;
    
    import java.io.IOException;
    import java.net.URI;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 8:19
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductDriver {
        public static void main(String[] args) throws Exception {
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            job.setJarByClass(OrderProductDriver.class);
    
            job.setMapperClass(OrderProductMapper.class);
    
            job.setOutputKeyClass(OrderProduct.class);
            job.setOutputValueClass(NullWritable.class);
    //缓存普通文件到task运行节点的目录去
            job.addCacheFile(new URI("file:/D:/datas/product.txt"));
            job.setNumReduceTasks(0);
    
    //        job.setOutputKeyClass(OrderProduct.class);
    //        job.setOutputValueClass(NullWritable.class);
    
            //job.addArchiveToClassPath();//缓存jar包到task运行节点的classpath中
            //job.addFileToClassPath();//缓存普通文件到task运行节点的classpath中
            //job.addCacheArchive();//缓存压缩包到task运行节点的目录中去
    
    
    
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
    
            job.waitForCompletion(true);
        }
    }

yarn上运行的代码：

package application.mapreduce;
    
    import org.apache.hadoop.fs.FileSystem;
    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.Mapper;
    
    import java.io.BufferedReader;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.URI;
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-11 7:46
     * @Version: 1.0
     * @Description:
     */
    public class OrderProductMapper extends Mapper<LongWritable, Text,OrderProduct, NullWritable> {
    
        private Text text = new Text();
        private Map<String,OrderProduct> orderProductMap = new HashMap<>();
    
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            URI[] caches = context.getCacheFiles();
            System.out.println(caches[0]);
            FileSystem fileSystem = FileSystem.get(context.getConfiguration());
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fileSystem.open(new Path(caches[0].getPath()))));
            String line = null;
            while((line = bufferedReader.readLine())!=null){
                String[] fields = line.split(" ");
                String pid = fields[0];
                String productName = fields[1];
                String categoryId = fields[2];
                double price = Double.parseDouble(fields[3]);
                OrderProduct orderProduct = OrderProduct.builder()
                        .date("")
                        .pid(pid)
                        .amount(0)
                        .productName(productName)
                        .categoryId(categoryId)
                        .price(price)
                        .flag(true)
                        .build();
                orderProductMap.put(pid,orderProduct);
            }
    
            System.out.println(orderProductMap);
        }
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split(" ");
            String date = fields[1];
            String pid = fields[2];
            int amount = Integer.parseInt(fields[3]);
            OrderProduct orderProduct = OrderProduct.builder()
                    .date(date)
                    .pid(pid)
                    .amount(amount)
                    .productName("")
                    .categoryId("")
                    .price(0)
                    .flag(false)
                    .build();
            orderProductMap.forEach((key1, value1) -> {
                if (key1.equalsIgnoreCase(orderProduct.getPid())){
                    OrderProduct orderProduct1 = OrderProduct.builder()
                            .date(orderProduct.getDate())
                            .pid(value1.getPid())
                            .amount(orderProduct.getAmount())
                            .productName(value1.getProductName())
                            .categoryId(value1.getCategoryId())
                            .price(value1.getPrice())
                            .build();
                    try {
                        context.write(orderProduct1,NullWritable.get());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70]: /images/20221123/4cb66f4f3ead4d4eadd22004ce982a7e.png