其实在实际的应用中在数据分析阶段可能会需要有序的数据，同时在hadoop和spark中值到reduce的顺序是未定义的，换句话就是没有明确的顺序，除非我们利用mapreduce的排序阶段（sort）将所需要的数据送达归约器（reduce），进而我们引入了反转排序【Order Inversion，OI】，用来控制mapreduce中的归约器值的顺序，OI模式比较适合成对模式，适当地确定提供reduce的数据顺序。来个栗子阐述下：
比如：计算给定文档集中单词的相对频度，为了更好理解我们通过矩阵来描述该问题：通过建一个 N x N的矩阵，其中N为所有给定文档的单词量，每个单元Mij则代表一个特定的邻域里单词Wi和Wj共同出现的次数【邻域可以简单理解为以某个单词为中心前后跨度单词的范围】，现有W1，W2，W3，W4，W5，W6 那么基于这个单词集合形成的邻域表如下,同时设定跨度为2
W1 W2，W3
W2 W1，W3，W4
W3 W1，W2, W4, W5
W4 W2，W3, W5, W6
W5 W3， W4, W6
W6 W4，W5
对应的计算方式：单词集合 tokens 变量i：代表单词集合元素下班 变量neighborWindow：每个单词的邻域跨度
// 循环遍历单词集合
for(int i=0; i<tokens.length;i++){
String word = tokens[i];
因为每一个单词的邻域范围[0,tokens.length)【主要因为数组下标从0开始故而上限不能超过tokens.length】
// 计算邻域范围的起点：通过将当前元素下标与跨度的差值是否>0;大于0说明该单词左侧存在其他单词，故而需要获取对应的下标（ i-neighborWindow）
//计算邻域范围的终点则比较当前单词的下标和跨度的和值是否>tokens.length;若是小于则说明该单词右边还有其他单词，故而获取对应的值(i+neighborWindow)
int start = (i-neighborWindow < 0) ? 0 : i-neighborWindow; 跨度起点
int end = (i+neighborWindow > tokens.length) ? tokens.length -1 : i+neighborWindow
}
上面的内容总结的结果：计算相对频度是需要得到边缘计算【领域范围的起点和终点】，在完成前面的需求，我们在得到所有的计算前，是无法完成边缘计算的，所以需要边缘计数在联合计数之前达到归约器【reduce】，需要注意一点计算相对频度尽量不要使用绝对单词数【绝对计算没有考虑某些单词可能比另外一些单词出现次数频繁，比如W1和W2经常出现，原因是W1比较常用】
N(Wi,Wj) 【单词对(Wi,Wj)出现的次数】
f(Wi|Wj) = -----------------
∑N(Wi,w)【Wi与所有的其他单词出现的总次数】
接下来我们将分别使用Hadoop和Spark分别实现反转排序
一、MapReduce实现
（1）、Driver
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
//
import org.apache.log4j.Logger;
public class RelativeFrequencyDriver
extends Configured implements Tool {
private static final Logger THE_LOGGER = Logger.getLogger(RelativeFrequencyDriver.class);
public static void main(String[] args) throws Exception {
if (args.length != 3) {
THE_LOGGER.warn("usage: <window> <input> <output>");
System.exit(-1);
}
int status = ToolRunner.run(new RelativeFrequencyDriver(), args);
System.exit(status);
}
@Override
public int run(String[] args) throws Exception {
int neighborWindow = Integer.parseInt(args[0]);
Path inputPath = new Path(args[1]);
Path outputPath = new Path(args[2]);
Job job = new Job(new Configuration(), "RelativeFrequencyDriver");
job.setJarByClass(RelativeFrequencyDriver.class);
job.setJobName("RelativeFrequencyDriver");
Delete the output directory if it exists already
FileSystem.get(getConf()).delete(outputPath, true);
job.getConfiguration().setInt("neighbor.window", neighborWindow);
FileInputFormat.setInputPaths(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
(key,value) generated by map()
job.setMapOutputKeyClass(PairOfWords.class);
job.setMapOutputValueClass(IntWritable.class);
(key,value) generated by reduce()
job.setOutputKeyClass(PairOfWords.class);
job.setOutputValueClass(DoubleWritable.class);
job.setMapperClass(RelativeFrequencyMapper.class);
job.setReducerClass(RelativeFrequencyReducer.class);
job.setCombinerClass(RelativeFrequencyCombiner.class);
job.setPartitionerClass(OrderInversionPartitioner.class);
job.setNumReduceTasks(3);
long startTime = System.currentTimeMillis();
job.waitForCompletion(true);
THE_LOGGER.info("Job Finished in milliseconds: " + (System.currentTimeMillis() - startTime));
return 0;
}
}
辅助参数： PairOfWords 组合键
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
//
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;
public class PairOfWords implements WritableComparable<PairOfWords> {
private String leftElement;
private String rightElement;
**
* Creates a pair.
*/
public PairOfWords() {
}
**
* Creates a pair.
*
* @param left the left element
* @param right the right element
*/
public PairOfWords(String left, String right) {
set(left, right);
}
**
* Deserializes the pair.
*
* @param in source for raw byte representation
*/
@Override
public void readFields(DataInput in) throws IOException {
leftElement = Text.readString(in);
rightElement = Text.readString(in);
}
**
* Serializes this pair.
*
* @param out where to write the raw byte representation
*/
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, leftElement);
Text.writeString(out, rightElement);
}
public void setLeftElement(String leftElement) {
this.leftElement = leftElement;
}
public void setWord(String leftElement) {
setLeftElement(leftElement);
}
**
* Returns the left element.
*
* @return the left element
*/
public String getWord() {
return leftElement;
}
**
* Returns the left element.
*
* @return the left element
*/
public String getLeftElement() {
return leftElement;
}
public void setRightElement(String rightElement) {
this.rightElement = rightElement;
}
public void setNeighbor(String rightElement) {
setRightElement(rightElement);
}
**
* Returns the right element.
*
* @return the right element
*/
public String getRightElement() {
return rightElement;
}
public String getNeighbor() {
return rightElement;
}
**
* Returns the key (left element).
*
* @return the key
*/
public String getKey() {
return leftElement;
}
**
* Returns the value (right element).
*
* @return the value
*/
public String getValue() {
return rightElement;
}
**
* Sets the right and left elements of this pair.
*
* @param left the left element
* @param right the right element
*/
public void set(String left, String right) {
leftElement = left;
rightElement = right;
}
/**
* Checks two pairs for equality.
*
* @param obj object for comparison
* @return <code>true</code> if <code>obj</code> is equal to this object, <code>false</code> otherwise
*/
@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
//
if (!(obj instanceof PairOfWords)) {
return false;
}
//
PairOfWords pair = (PairOfWords) obj;
return leftElement.equals(pair.getLeftElement())
&& rightElement.equals(pair.getRightElement());
}
/**
* Defines a natural sort order for pairs. Pairs are sorted first by the left element, and then by the right
* element.
*
* @return a value less than zero, a value greater than zero, or zero if this pair should be sorted before, sorted
* after, or is equal to <code>obj</code>.
*/
@Override
public int compareTo(PairOfWords pair) {
String pl = pair.getLeftElement();
String pr = pair.getRightElement();
if (leftElement.equals(pl)) {
return rightElement.compareTo(pr);
}
return leftElement.compareTo(pl);
}
/**
* Returns a hash code value for the pair.
*
* @return hash code for the pair
*/
@Override
public int hashCode() {
return leftElement.hashCode() + rightElement.hashCode();
}
/**
* Generates human-readable String representation of this pair.
*
* @return human-readable String representation of this pair
*/
@Override
public String toString() {
return "(" + leftElement + ", " + rightElement + ")";
}
/**
* Clones this object.
*
* @return clone of this object
*/
@Override
public PairOfWords clone() {
return new PairOfWords(this.leftElement, this.rightElement);
}
/**
* Comparator optimized for <code>PairOfWords</code>.
*/
public static class Comparator extends WritableComparator {
/**
* Creates a new Comparator optimized for <code>PairOfWords</code>.
*/
public Comparator() {
super(PairOfWords.class);
}
/**
* Optimization hook.
*/
@Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
try {
int firstVIntL1 = WritableUtils.decodeVIntSize(b1[s1]);
int firstVIntL2 = WritableUtils.decodeVIntSize(b2[s2]);
int firstStrL1 = readVInt(b1, s1);
int firstStrL2 = readVInt(b2, s2);
int cmp = compareBytes(b1, s1 + firstVIntL1, firstStrL1, b2, s2 + firstVIntL2, firstStrL2);
if (cmp != 0) {
return cmp;
}
int secondVIntL1 = WritableUtils.decodeVIntSize(b1[s1 + firstVIntL1 + firstStrL1]);
int secondVIntL2 = WritableUtils.decodeVIntSize(b2[s2 + firstVIntL2 + firstStrL2]);
int secondStrL1 = readVInt(b1, s1 + firstVIntL1 + firstStrL1);
int secondStrL2 = readVInt(b2, s2 + firstVIntL2 + firstStrL2);
return compareBytes(b1, s1 + firstVIntL1 + firstStrL1 + secondVIntL1, secondStrL1, b2,
s2 + firstVIntL2 + firstStrL2 + secondVIntL2, secondStrL2);
} catch (IOException e) {
throw new IllegalArgumentException(e);
}
}
}
static { // register this comparator
WritableComparator.define(PairOfWords.class, new Comparator());
}
}
（2）、自定义partition ：由于这里面我们需要使用组合键的方式(W,Wj)【代表和单词W公共出现的单词组合，一般W是不变的】，为了能够使组合键(W,Wi)左边的W投递到同一个reduce上面，需要定制分区器
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
public class OrderInversionPartitioner
extends Partitioner<PairOfWords, IntWritable> {
@Override
public int getPartition(PairOfWords key,
IntWritable value,
int numberOfPartitions) {
// key = (leftWord, rightWord) = (word, neighbor)
String leftWord = key.getLeftElement();
return Math.abs( ((int) hash(leftWord)) % numberOfPartitions);
}
/**
* 重载hashcode的计算
*/
private static long hash(String str) {
long h = 1125899906842597L; // prime
int length = str.length();
for (int i = 0; i < length; i++) {
h = 31*h + str.charAt(i);
}
return h;
}
}
（3）、Mapper实现
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
//
import org.apache.commons.lang.StringUtils;
//
import java.io.IOException;
public class RelativeFrequencyMapper
extends Mapper<LongWritable, Text, PairOfWords, IntWritable> {
private int neighborWindow = 2;
// pair = (leftElement, rightElement)
private final PairOfWords pair = new PairOfWords();
private final IntWritable totalCount = new IntWritable();
private static final IntWritable ONE = new IntWritable(1);
@Override
public void setup(Context context) {
this.neighborWindow = context.getConfiguration().getInt("neighbor.window", 2);
}
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] tokens = StringUtils.split(value.toString(), " ");
//String[] tokens = StringUtils.split(value.toString(), "\\s+");
if ((tokens == null) || (tokens.length < 2)) {
return;
}
for (int i = 0; i < tokens.length; i++) {
tokens[i] = tokens[i].replaceAll("\\W+", "");
if (tokens[i].equals("")) {
continue;
}
pair.setWord(tokens[i]);
// 计算邻域的边缘计数
int start = (i - neighborWindow < 0) ? 0 : i - neighborWindow;
int end = (i + neighborWindow >= tokens.length) ? tokens.length - 1 : i + neighborWindow;
for (int j = start; j <= end; j++) {
if (j == i) {
continue;
}
pair.setNeighbor(tokens[j].replaceAll("\\W", ""));
context.write(pair, ONE);
}
//
pair.setNeighbor("*");
totalCount.set(end - start);
context.write(pair, totalCount);
}
}
}
