您可以在hadoop分布式缓存中加载术语索引列表，以便Map器和还原器可以使用它。例如，在hadoop流媒体中，可以按如下方式运行作业：

$ hadoop  jar $HADOOP_INSTALL/contrib/streaming/hadoop-streaming-*.jar \
  -input myInputDirs \
  -output myOutputDir \
  -mapper myMapper.py \
  -reducer myReducer.py \
  -file myMapper.py \
  -file myReducer.py \
  -file myTermIndexList.txt

现在，在mymapper.py中，您可以加载mytermindexlist.txt文件并根据需要使用它。如果你对你的输入和期望的输出有更详细的描述，我可以给你更多的细节。

好吧，你可以使用类似于倒排索引的概念。
我建议这样做是因为，我假设两个文件都很大。因此，像一对一那样相互比较将是真正的性能瓶颈。
这里有一种方法-
您可以将输入文件格式csv文件（例如datafile1、datafile2）和术语索引文件（例如term\u index\u file）作为输入提供给作业。
然后在每个Map器中，过滤源文件名，类似这样-
Map器的伪代码-

map(key, row, context){
  String filename= ((FileSplit)context.getInputSplit()).getPath().getName();
   if (filename.startsWith("datafile") {

     //split the review_id, words from row
     ....
     context.write(new Text("word), new Text("-1 | review_id"));

   } else if(filename.startsWith("term_index_file") {
     //split index and word
     ....
     context.write(new Text("word"), new Text("index | 0"));
   }

}

e、 g.不同Map器的输出

Key       Value      source
product   -1|1       datafile
very       5|0       term_index_file
very      -1|1       datafile
product   -1|2       datafile
very      -1|1       datafile
product    2|0       term_index_file
...
...

解释（示例）：因为它清楚地显示了关键字将是您的单词，值将由两部分组成，两部分之间用分隔符“|”分隔
如果源是一个数据文件，则发出key=product和value=-1 | 1，其中-1是一个伪元素，1是一个review | id。
如果源是一个term|index|文件，则发出key=product和value=2 | 0，其中2是单词“product”的索引，0是一个伪review|id，我们将使用它进行排序-稍后解释。
当然，没有重复的索引将由两个不同的Map器处理，如果我们是提供一个正常的输入文件的作业的长期索引文件。因此，“product，vary”或术语索引文件中的任何其他索引词将仅对一个Map器可用。注意这只对term\索引\文件有效，而不是数据文件。
下一步：
hadoop mapreduce框架，如你所知，将按键分组，所以，你将有这样的东西去不同的还原器，

reduce-1: key=product, value=<-1|1, -1|2, 2|0>
reduce-2: key=very, value=<5|0, -1|1, -1|1>

但是，在上述情况下，我们有一个问题。我们希望在“|”之后的值中进行排序，即 reduce-1 -> 2|0, -1|1, -1|2 and in reduce-2 -> <5|0, -1|1, -1|1> 为了实现这一点，您可以使用使用排序比较器实现的辅助排序。请用谷歌搜索，但这里的链接可能会有所帮助。在这里提到它可能会很长时间。
在每个reduce-1中，由于值是按上述方式排序的，因此当我们开始迭代时，我们将在第一次迭代中得到“0”，并用它得到索引\u id=2，然后可以用于后续迭代。在接下来的两次迭代中，我们连续地得到review id 1和2，并使用一个计数器，以便跟踪任何重复的review id。当我们得到重复的评论id时，这意味着一个单词在同一个评论id行中出现了两次。只有在找到不同的审阅id并发出特定索引id的上一个审阅id详细信息时，我们才重置计数器，类似这样-

previous_review_id  + "\t" + index_id + "\t" + count

当循环结束时，我们将只剩下一个先前的\u review \u id，我们最终以相同的方式发出它。
减速机伪代码-

reduce(key, Iterable values, context) {
  String index_id = null;
  count = 1;
  String previousReview_id = null;
  for(value: values) {
      Split split[] = values.split("\\|");
      ....

      //when consecutive review_ids are same, we increment count
      //and as soon as the review_id differ, we emit, reset the counter and print
      //the previous review_id detected.
      if (split[0].equals("-1") && split[1].equals(previousReview_id)) {
          count++;
      } else if(split[0].equals("-1") && !split[1].equals(prevValue)) {
          context.write(previousReview_id + "\t" + index_id + "\t" + count);
          previousReview_id = split[1];//resting with new review_id id
          count=1;//resetting count for new review_id
      } else {
         index_id = split[0]; 
      }
  }
  //the last  previousReview_id will be left out, 
  //so, writing it now after the loop  completion
  context.write(previousReview_id + "\t" + index_id + "\t" + count);

}

这项工作是用多个reducer来完成的，目的是利用hadoop实现它最著名的性能，结果，最终的输出将分散，类似下面这样，偏离您期望的输出。

1 4 1
2 1 1
1 5 2
1 2 1
1 3 1
1 6 1
2 2 1

但是，如果您想根据review\u id（作为您想要的输出）对所有内容进行排序，您可以使用一个reducer并将previos作业的输出作为输入，为您编写另一个作业。同时计算2 6 7并将其放在输出的前面。
我认为，这只是一种方法（或想法），可能会对你有所帮助。你肯定想修改这个，把一个更好的算法，并使用它的方式，你认为会有利于你。
也可以使用复合键，以比使用分隔符（如“|”）更清晰。
我愿意澄清。如果你认为它可能对你有用，请询问。
谢谢您！

方法#1并不好，但是如果你没有太多的hadoop经验的话，这种方法非常常见。开始工作是非常昂贵的。你要做的是做2-3份工作，互相补充，以得到想要的结果。类似问题的一个常见解决方案是让Map器对输入和输出对进行标记化，将它们分组到执行某种计算的reducer中，然后将其输入到job 2中。在job 2中的mapper中，以某种方式反转数据，并在reducer中执行其他计算。
我强烈建议通过培训课程学习更多关于hadoop的知识。有趣的是，cloudera的开发课程有一个与您试图解决的问题非常相似的问题。或者除了一门课程之外，我还将学习“使用mapreduce进行数据密集型文本处理”，特别是“计算相对频率”和“文本检索的反向索引”部分
http://lintool.github.io/mapreducealgorithms/mapreduce-book-final.pdf