MapRed v2

// https://www.edureka.co/blog/mapreduce-example-reduce-side-join/
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class custTrans {

    public static class mapCustDetails extends Mapper<Object, Text, Text, Text> {

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String record = value.toString();
            String[] parts = record.split(",");
            context.write(new Text(parts[0]), new Text("cust\t" + parts[1]));
        }
    }

    public static class mapTransDetails extends Mapper<Object, Text, Text, Text> {

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String record = value.toString();
            String[] parts = record.split(",");
            context.write(new Text(parts[2]), new Text("trans\t" + parts[3]));
        }
    }

    public static class transReducer extends Reducer<Text, Text, Text, Text> {

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            String name = "";
            int count = 0;
            double amount = 0.0;

            for(Text value : values) {
                String parts[] = value.toString().split("\t");

                if(parts[0].equals("trans")) {
                    count++;
                    amount += Float.parseFloat(parts[1]);
                } else if(parts[0].equals("cust")) {
                    name = parts[1];
                }

                String str = String.format("%d\t%f", count, amount);
                context.write(new Text(name), new Text(str));
            }
        }
    }

    @SuppressWarnings("deprecation")
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration con = new Configuration();

        Job job = new Job(con, "Sports Transactions");

        job.setJarByClass(custTrans.class);
        job.setReducerClass(transReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, mapCustDetails.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, mapTransDetails.class);

        Path outputPath = new Path(args[2]);

        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(con).delete(outputPath);
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}