@Data
public class OrderBean implements Writable {
    private String orderId = "";
    private String pid = "";
    private int amount;
    private String pname = "";
    /**
     * 用来区分是order表的封装对象还是product表的封装对象
     */
    private String flag = "";

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(orderId);
        dataOutput.writeUTF(pid);
        dataOutput.writeInt(amount);
        dataOutput.writeUTF(pname);
        dataOutput.writeUTF(flag);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.orderId = dataInput.readUTF();
        this.pid = dataInput.readUTF();
        this.amount = dataInput.readInt();
        this.pname = dataInput.readUTF();
        this.flag = dataInput.readUTF();
    }

    @Override
    public String toString() {
        return orderId + " " + pname + " " + amount;
    }
}

public class OrderMapper extends Mapper<LongWritable, Text, Text, OrderBean> {
    /**
     * 当前切片所在文件名字
     */
    String fileName;
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 获得切片
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        // 获取切片数据所在文件路径
        Path path = fileSplit.getPath();
        // 获取文件名
         fileName = path.getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] array = line.split("\t");
        OrderBean ob = new OrderBean();
        String file = "order";
        if (fileName.contains(file)) {
            // 如果是订单表，只需要向OrderBean对象中封装：orderId、pid、amount、flag
            ob.setOrderId(array[0]);
            ob.setPid(array[1]);
            ob.setAmount(Integer.parseInt(array[2]));
            ob.setFlag("order");
        } else {
            // 如果是产品表，只需要向OrderBean对象中封装：pid、flag
            ob.setPid(array[0]);
            ob.setPname(array[1]);
            ob.setFlag("product");
        }
        context.write(new Text(ob.getPid()), ob);
    }
}

public class OrderReducer extends Reducer<Text, OrderBean, NullWritable, OrderBean> {
    /**
     * @param key   产品id
     * @param values   产品id相同的订单信息和产品信息
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<OrderBean> values, Context context) throws IOException, InterruptedException {
        // 订单有多个
        List<OrderBean> orders = new ArrayList<>();
        // 产品信息只有一个
        OrderBean product = new OrderBean();
        /**
         * 在使用增强的for循环时，前面的变量永远都是同一个对象，只不过指向的地址不一样了。
         * 如果直接把value的值赋给其他变量，那么value的地址值变化，那么赋值给的变量也就发生变化了
         */
        for (OrderBean value: values) {
            // 如果是订单数据，将订单数据加到集合中
            if ("order".equals(value.getFlag())) {
                OrderBean orderBean = new OrderBean();
                // 将value中的数据copy到orderBean中。增强的for循环中每一次获取的对象都复制给同一个值，如果加到集合中只能加上一个，所以需要通过这种方式将两个对象区分开
                orderBean.setOrderId(value.getOrderId());
                orderBean.setPid(value.getPid());
                orderBean.setAmount(value.getAmount());
                orderBean.setFlag(value.getFlag());
                orders.add(orderBean);
            } else {
                // 如果是产品信息，那么将产品信息数据赋值给product对象
                product.setPid(value.getPid());
                product.setPname(value.getPname());
                product.setFlag(value.getFlag());
            }
        }

        // 将订单中的产品名字修改为产品信息中的名字
        for (OrderBean order: orders) {
            order.setPname(product.getPname());
            context.write(NullWritable.get(), order);
        }
    }
}

public class OrderDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.218.55:9000");
        Job job = Job.getInstance(conf);

        job.setJarByClass(OrderDriver.class);
        job.setMapperClass(OrderMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(OrderBean.class);

        job.setReducerClass(OrderReducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(OrderBean.class);

        FileInputFormat.setInputPaths(job, new Path("/school/join/*"));
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");
        Path output = new Path("/school/join/test_reduce");
        if (fs.exists(output)) {
            fs.delete(output, true);
        }
        FileOutputFormat.setOutputPath(job, output);

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

public class OrderDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.218.55:9000");
        Job job = Job.getInstance(conf);

        job.setJarByClass(OrderDriver.class);
        job.setMapperClass(OrderMapper.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        job.setNumReduceTasks(0);

        // 需要把小文件提前缓存下来-----需要在map阶段的setup方法中读取缓存进行数据处理
        job.addCacheFile(new URI("/school/join/product.txt"));

        // 输入文件只需要指定大文件，即订单文件的输入路径即可
        FileInputFormat.setInputPaths(job, new Path("/school/join/order.txt"));
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");
        Path output = new Path("/school/join/test_map");
        if (fs.exists(output)) {
            fs.delete(output, true);
        }
        FileOutputFormat.setOutputPath(job, output);

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

public class OrderMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
    Map<String, String> products = new HashMap<>();
    /**
     * 缓存的小文件数据读取、存储集合备用。
     * 小文件是产品表，一个字段是pid，一个字段是pname
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 获取缓存的小文件
        URI[] cacheFiles = context.getCacheFiles();
        URI cacheFile = cacheFiles[0];
        // 获取缓存文件的路径
        String path = cacheFile.getPath();
        try {
            // 创建一个文件系统，用来获取缓存文件的io流
            FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), context.getConfiguration(), "root");
            // 获取缓存文件的io流
            FSDataInputStream open = fs.open(new Path(path));
            // 将io流转换为字符缓冲流，可以实现一次读取文件的一行数据
            BufferedReader br = new BufferedReader(new InputStreamReader(open));
            String line = null;
            while ((line = br.readLine()) != null) {
                // 把一行数据切割，得到[1， 小米]
                String[] array = line.split("\t");
                // 将数据传到集合中
                products.put(array[0], array[1]);
            }
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    /**
     * map只需要处理订单表的数据即可
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // order.txt中的一行数据 [1001, 01, 5]
        String[] array = value.toString().split("\t");
        String orderId = array[0];
        String pid = array[1];
        String amount = array[2];

        // 获取产品名---根据订单文件的pid去刚刚缓存下来的产品集合中获取值即可
        String pname = products.get(pid);
        String line = orderId + "\t" + pname + "\t" + amount;
        context.write(NullWritable.get(), new Text(line));
    }
}

public class SchoolDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.218.55:9000");
        Job job = Job.getInstance(conf);

        job.setJarByClass(SchoolDriver.class);
        job.setMapperClass(SchoolMapper.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        job.addCacheFile(new URI("/school/stu/class.txt"));
        job.addCacheFile(new URI("/school/stu/dept.txt"));

        FileInputFormat.setInputPaths(job, new Path("/school/stu/student.txt"));
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");
        Path path = new Path("/test/school/school_map");
        if (fs.exists(path)) {
            fs.delete(path, true);
        }
        FileOutputFormat.setOutputPath(job, path);

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);

    }
}

public class SchoolMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
    Map<String, String> dept = new HashMap<>();
    Map<String, String[]> cla = new HashMap<>();
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] cacheFiles = context.getCacheFiles();
        URI file1 = cacheFiles[0];
        URI file2 = cacheFiles[1];
        String path1 = file1.getPath();
        String path2 = file2.getPath();
        FileSystem fs = FileSystem.get(context.getConfiguration());
        FSDataInputStream open1 = fs.open(new Path(path1));
        FSDataInputStream open2 = fs.open(new Path(path2));
        BufferedReader br1 = new BufferedReader(new InputStreamReader(open1));
        BufferedReader br2 = new BufferedReader(new InputStreamReader(open2));
        String line = null;
        while ((line = br1.readLine()) != null) {
            String[] split = line.split("\t");
            cla.put(split[0], new String[]{split[1], split[2]});
        }
        while ((line = br2.readLine()) != null) {
            String[] split = line.split("\t");
            dept.put(split[0], split[1]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] split = line.split("\t");
        String[] classInfo = cla.get(split[1]);
        String sid = split[0];
        String className = classInfo[1];
        String deptName = dept.get(classInfo[0]);
        String score = split[2];

        String result = sid + "\t" + className + "\t" + deptName + "\t" + score;
        context.write(NullWritable.get(), new Text(result));
    }
}