看起来像是一个常规的reduce-side连接，所以它可能是一个简单的用例，然而这些类型的计算往往在工作负载方面变得非常残酷。这意味着我们必须找到捷径，以确保应用程序能够很好地扩展到更大的输入规模。
为应用程序的执行节省时间/空间的最常见的方法是，尝试设计可能的几个mr作业，使我们可以在保留所有功能的同时“剪切”一个或多个作业，或者尝试最小化将在输入数据处实现的（自定义）Map器的数量。这两个函数中的后一个对于您正在尝试实现的这种过滤非常常见，因为我们可以很容易地使用一个map函数，它的每个示例将检查当前正在读取的文件的名称，从而相应地执行操作。
更具体地说，我们可以得到 File1.txt 以及 File2.txt 在Map程序开始遍历 setup 函数，并使用当前要读取的文件名来确定如何将文件中的数据切分并存储到键值对中。对于您的问题，此Map函数将输出两种类型的键值对： <customer_ID, customer_name> （对于中的数据） File1.txt ) <customer_ID, order_ID> （对于中的数据） File2.txt )
然后reduce函数的示例将为每个客户运行（当然，因为客户id和名称是唯一的），并访问分组的值，这些值只不过是一些 Text 包含此客户名称或订单id的对象。我们只想输出记录中没有任何订单的客户，因此我们所要做的就是检查此值列表的长度是否为 1 （也就是说，如果这个客户没有一对值，而另一个客户没有他的名字）。
为了展示这一点，我将两个输入文件放在一个目录中 /input 在hdfs中（我对中的列使用了两个制表符分隔符） File1.txt 和三个制表符分隔符 File2.txt . 如果您的文件在列之间有不同的选项卡或空格，您可以相应地更改它们）：
文件1.txt

Cust.No Name
1       Adam
2       Abe
3       Alex
4       Jones

文件2.txt

Order.Num.  Cust.No.    Price
01          1           5422
02          1           23
03          2           1265
04          3           127

执行过滤的程序可以如下所示：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class OrderListFilter
{
    /* input:  <byte_offset, line_of_dataset>
     * output: <customer_ID, customer_name> OR <customer_ID, order_ID>
     */
    public static class Map extends Mapper<LongWritable, Text, Text, Text>
    {
        private String current_filename = "";

        protected void setup(Context context)
        {
            // get the name of the current to-be-read file
            InputSplit split = context.getInputSplit();
            Path path = ((FileSplit) split).getPath();
            current_filename = path.getName();
        }

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
        {
            if(current_filename.equals("File1.txt"))    // if mapper is reading through the customer's file
            {
                if(value.toString().contains("Cust.No"))    // remove header
                    return;
                else
                {
                    String[] columns = value.toString().split("\t\t");  // 2 tabs as delimiter

                    // write customer ID as key and name as value
                    context.write(new Text(columns[0]), new Text(columns[1]));
                }
            }
            else if(current_filename.equals("File2.txt"))   // if mapper is reading through the order's file
            {
                if(value.toString().contains("Cust.No"))    // remove header
                    return;
                else
                {
                    String[] columns = value.toString().split("\t\t\t"); // 3 tabs as delimiter

                    // write customer ID as key and order num as value
                    context.write(new Text(columns[1]), new Text(columns[0]));
                }
            }
        }
    }

    /* input: <customer_ID, customer_name> OR <customer_ID, order_ID>
     * output: <customer_ID, customer_name>
     */
    public static class Reduce extends Reducer<Text, Text, Text, Text>
    {
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
        {
            List<String> customer_records = new ArrayList<String>();

            // put all the values in a list to find the size of them
            for(Text value : values)
                customer_records.add(value.toString());

            // if there's only one record, i.e. just the ID and the customer's name in they key-value pairs,
            // write their ID and name to output
            if(customer_records.size() == 1)
                context.write(key, new Text(customer_records.get(0)));
        }
    }

    public static void main(String[] args) throws Exception
    {
        // set the paths of the input and output directories in the HDFS
        Path input_dir = new Path("input");
        Path output_dir = new Path("output");

        // in case the output directory already exists, delete it
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        if(fs.exists(output_dir))
            fs.delete(output_dir, true);

        // configure the MapReduce job
        Job job = Job.getInstance(conf, "Order List Filter");
        job.setJarByClass(OrderListFilter.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, input_dir);
        FileOutputFormat.setOutputPath(job, output_dir);
        job.waitForCompletion(true);
    }
}

它的输出看起来是一个好的（忽略我的设置中的警告）：