代码-JZTXT

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

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

private final static LongWritable one = new LongWritable(1);  

private Text word = new Text();  



@Override  

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  

    String line = value.toString();  

    String[] words = line.split(" ");  

    for (String w : words) {  

        word.set(w);  

        context.write(word, one);  

    }  

}

}
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer<Text, Text, Text, Text> {
private Text result = new Text();

@Override  
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {  
    for (Text val : values) {  
        result.set(key + " " + val);  
        context.write(result, null);  
    }  
}

}
// Map class
public class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
private Text stockName = new Text();
private IntWritable stockPrice = new IntWritable();

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
    String[] line = value.toString().split(" ");  
    if (line[1].equals("Buy")) {  
        stockName.set(line[0]);  
        stockPrice.set(-Integer.parseInt(line[2]));  
        context.write(stockName, stockPrice);  
    } else if (line[1].equals("Sell")) {  
        stockName.set(line[0]);  
        stockPrice.set(Integer.parseInt(line[2]));  
        context.write(stockName, stockPrice);  
    }  
}

}

// Reduce class
public class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
int total = 0;
for (IntWritable val : values) {
total += val.get();
}
context.write(key, new IntWritable(total));
}
}

// Driver class
public class StockCapitalGain {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Stock Capital Gain");
job.setJarByClass(StockCapitalGain.class);
job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
public class FindMutualFollowersMapper extends Mapper<LongWritable, Text, Text, Text> {
private Text user = new Text();
private Text following = new Text();

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
    String[] data = value.toString().split(",");  
    String userString = data[1];  
    List<String> followingList = Arrays.asList(data[2].split(","));  
    for (String follower : followingList) {  
        user.set(userString);  
        following.set(follower);  
        context.write(new Text(userString + "<->" + follower), null);  
    }  
}

}
public class FindMutualFollowersReducer extends Reducer<Text, Text, Text, Text> {
public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
for (Text val : values) {
context.write(key, val);
}
}
}