测试数据:

job-liu

fei-hh

hh-uu

qq-ww

ee-bb

bb-yy

package com.mp;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class MapCal extends Mapper<LongWritable, Text, Text, Text> {

@Override

protected void map(LongWritable lon, Text value, Context context)

throws IOException, InterruptedException {

String line = value.toString();

String[] peps = line.split("-");

// 键值对

context.write(new Text(peps[0]), new Text("s" + peps[1]));

context.write(new Text(peps[1]), new Text("g" + peps[0]));

}

}

public class ReduceCal extends Reducer<Text, Text, Text, Text> {

@Override

protected void reduce(Text arg0, Iterable<Text> arg1, Context context)

throws IOException, InterruptedException {

ArrayList<Text> grands = new ArrayList<Text>();

ArrayList<Text> sons = new ArrayList<Text>();

// 把这些值写入集合

for (Text text : arg1) {

String str = text.toString();

if (str.startsWith("g")) {

grands.add(new Text(str.subString(1)));

} else {

sons.add(new Text(str.subString(1)));

}

}

// 输出

for (int i = 0; i < grands.size(); i++) {

for (int j = 0; j < sons.size(); j++) {

context.write(grands.get(i), sons.get(j));

}

}

}

}

package com.mp;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class JobRun {

public static void main(String[] args) {

try {

Configuration conf = new Configuration();

FileSystem fs = FileSystem.get(conf);

Job job = Job.getInstance(conf);

job.setJobName("jc11");

job.setJarByClass(JobRun.class);

job.setMapperClass(MapCal.class);

job.setReducerClass(ReduceCal.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(Text.class);

FileInputFormat.addInputPath(job, new Path("/shuju.txt"));

if (fs.exists(new Path("/outg"))) {

fs.delete(new Path("/outg"));

}

FileOutputFormat.setOutputPath(job, new Path("/outg"));

boolean f = job.waitForCompletion(true);

if (f) {

System.out.println("ok");

}

} catch (Exception e) {

e.printStackTrace();

}

}

}

Shuffle Errors

BAD_ID=0

CONNECTION=0

IO_ERROR=0

WRONG_LENGTH=0

WRONG_MAP=0

WRONG_REDUCE=0

File Input Format Counters 

Bytes Read=45

File Output Format Counters 

Bytes Written=18

eeyy

feiuu