import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
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.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class WordCount {
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
// 开始 Map 过程
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
//遍历 Map 里面的字符串
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
//开始 Reduce 过程
public void reduce(Text key, Iterable<IntWritable> values,Context context)
 throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
} 
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: wordcount <in> <out>");
System.exit(2);
}
Job job = new Job(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
 JavaSparkContext ctx = new JavaSparkContext(sparkConf);
 JavaRDD<String> lines = ctx.textFile(args[0], Integer.parseInt(args[1]));
 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterable<String> call(String s) {
 return Arrays.asList(SPACE.split(s));
 }
 });
 //定义 RDD ones
 JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(String s) {
 return new Tuple2<String, Integer>(s, 1);
 }
 });
 //ones.reduceByKey(func, numPartitions)
 JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer i1, Integer i2) {
 return i1 + i2;
 }
 },10);
//输出 List
List<Tuple2<String, Integer>> output = counts.collect();
 Collections.sort(output, new Comparator<Tuple2<String, Integer>>() {
 @Override
 public int compare(Tuple2<String, Integer> t1,
Tuple2<String, Integer> t2) {
if(t1._2 > t2._2) {
 return -1;
 } else if(t1._2 < t2._2) {
 return 1;
 }
 return 0;
}
 });

def fun_map(line, params):
 for word in line.split():
 yield word, 1

def fun_reduce(iter, params):
 from disco.util import kvgroup
 for word, counts in kvgroup(sorted(iter)):
 yield word, sum(counts)

from disco.core import Job, result_iterator
def map(line, params):
 for word in line.split():
 yield word, 1
def reduce(iter, params):
 from disco.util import kvgroup
 for word, counts in kvgroup(sorted(iter)):
 yield word, sum(counts)
if __name__ == '__main__':
 job = Job().run(input=["http://discoproject.org/media/text/chekhov.txt"],
 map=map,
 reduce=reduce)
 for word, count in result_iterator(job.wait(show=True)):
 print(word, count)
Note

#include <stdio.h>
#include <strings.h>
#include <string.h>
#include <stddef.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <fcntl.h>
#include <ctype.h>
#include <inttypes.h>
#include "map_reduce.h"
#include "stddefines.h"
#include "sort.h"
#define DEFAULT_DISP_NUM 10
typedef struct {
 int fpos;
 off_t flen;
 char *fdata;
 int unit_size;
} wc_data_t;
enum {
 IN_WORD,
 NOT_IN_WORD
};
 struct timeval begin, end;
#ifdef TIMING
 unsigned int library_time = 0;
#endif
/** mystrcmp()
 * Comparison function to compare 2 words
 */
int mystrcmp(const void *s1, const void *s2)
{
 return strcmp((const char *)s1, (const char *) s2);
}
/** mykeyvalcmp()
 * Comparison function to compare 2 ints
 */
int mykeyvalcmp(const void *v1, const void *v2)
{
 keyval_t* kv1 = (keyval_t*)v1;
 keyval_t* kv2 = (keyval_t*)v2;
 intptr_t *i1 = kv1->val;
 intptr_t *i2 = kv2->val;
 if (i1 < i2) return 1;
 else if (i1 > i2) return -1;
 else {
 return strcmp((char *)kv1->key, (char *)kv2->key);
 //return 0;
 }
}
/** wordcount_分割器 ()
 * 内存里面进行 Map 计算
 */
int wordcount_splitter(void *data_in, int req_units, map_args_t *out)
{
 wc_data_t * data = (wc_data_t *)data_in; 
 assert(data_in);
 assert(out);
 assert(data->flen >= 0);
 assert(data->fdata);
 assert(req_units);
 assert(data->fpos >= 0);
 // End of file reached, return FALSE for no more data
 if (data->fpos >= data->flen) return 0;
 // Set the start of the next data
 out->data = (void *)&data->fdata[data->fpos];
 // Determine the nominal length
 out->length = req_units * data->unit_size;
 if (data->fpos + out->length > data->flen)
 out->length = data->flen - data->fpos;
 // Set the length to end at a space
 for (data->fpos += (long)out->length;
 data->fpos < data->flen && 
 data->fdata[data->fpos] != ' ' && data->fdata[data->fpos] != '\t' &&
 data->fdata[data->fpos] != '\r' && data->fdata[data->fpos] != '\n';
 data->fpos++, out->length++);
 return 1;
}
/** wordcount_locator()
 * Return the memory address where this map task would heavily access.
 */
void *wordcount_locator (map_args_t *task)
{
 assert (task);
 return task->data;
}
/** wordcount_map()
 * 对文本进行计数
 */
void wordcount_map(map_args_t *args) 
{
 char *curr_start, curr_ltr;
 int state = NOT_IN_WORD;
 int i;
 assert(args);
 char *data = (char *)args->data;
 assert(data);
 curr_start = data;
 for (i = 0; i < args->length; i++)
 {
 curr_ltr = toupper(data[i]);
 switch (state)
 {
 case IN_WORD:
 data[i] = curr_ltr;
 if ((curr_ltr < 'A' || curr_ltr > 'Z') && curr_ltr != '\'')
 {
 data[i] = 0;
 emit_intermediate(curr_start, (void *)1, &data[i] - curr_start + 1);
 state = NOT_IN_WORD;
 }
 break;
 default:
 case NOT_IN_WORD:
 if (curr_ltr >= 'A' && curr_ltr <= 'Z')
 {
 curr_start = &data[i];
 data[i] = curr_ltr;
 state = IN_WORD;
 }
 break;
 }
 }
 // Add the last word
 if (state == IN_WORD)
 {
 data[args->length] = 0;
 emit_intermediate(curr_start, (void *)1, &data[i] - curr_start + 1);
 }
}
/** wordcount_reduce()
 * 计算字符
 */
void wordcount_reduce(void *key_in, iterator_t *itr)
{
 char *key = (char *)key_in;
 void *val;
 intptr_t sum = 0;
 assert(key);
 assert(itr);
 while (iter_next (itr, &val))
 {
 sum += (intptr_t)val;
 }
 emit(key, (void *)sum);
}
void *wordcount_combiner (iterator_t *itr)
{
 void *val;
 intptr_t sum = 0;
 assert(itr);
 while (iter_next (itr, &val))
 {
 sum += (intptr_t)val;
 }
 return (void *)sum;
}
int main(int argc, char *argv[]) 
{
 final_data_t wc_vals;
 int i;
 int fd;
 char * fdata;
 int disp_num;
 struct stat finfo;
 char * fname, * disp_num_str;
 struct timeval starttime,endtime;
 get_time (&begin);
 // 确保文件名
 if (argv[1] == NULL)
 {
 printf("USAGE: %s <filename> [Top # of results to display]\n", argv[0]);
 exit(1);
 }
 fname = argv[1];
 disp_num_str = argv[2];
 printf("Wordcount: Running...\n");
 // 读取文件
 CHECK_ERROR((fd = open(fname, O_RDONLY)) < 0);
 // Get the file info (for file length)
 CHECK_ERROR(fstat(fd, &finfo) < 0);
#ifndef NO_MMAP
 // 内存里面开始调用 map
 CHECK_ERROR((fdata = mmap(0, finfo.st_size + 1, 
 PROT_READ | PROT_WRITE, MAP_PRIVATE, fd, 0)) == NULL);
#else
 int ret;
 fdata = (char *)malloc (finfo.st_size);
 CHECK_ERROR (fdata == NULL);
 ret = read (fd, fdata, finfo.st_size);
 CHECK_ERROR (ret != finfo.st_size);
#endif
 CHECK_ERROR((disp_num = (disp_num_str == NULL) ? 
 DEFAULT_DISP_NUM : atoi(disp_num_str)) <= 0);
 wc_data_t wc_data;
 wc_data.unit_size = 5; // approx 5 bytes per word
 wc_data.fpos = 0;
 wc_data.flen = finfo.st_size;
 wc_data.fdata = fdata;
 CHECK_ERROR (map_reduce_init ());
 map_reduce_args_t map_reduce_args;
 memset(&map_reduce_args, 0, sizeof(map_reduce_args_t));
 map_reduce_args.task_data = &wc_data;
 map_reduce_args.map = wordcount_map;
 map_reduce_args.reduce = wordcount_reduce;
 map_reduce_args.combiner = wordcount_combiner;
 map_reduce_args.splitter = wordcount_splitter;
 map_reduce_args.locator = wordcount_locator;
 map_reduce_args.key_cmp = mystrcmp;
 map_reduce_args.unit_size = wc_data.unit_size;
 map_reduce_args.partition = NULL; // use default
 map_reduce_args.result = &wc_vals;
 map_reduce_args.data_size = finfo.st_size;
 map_reduce_args.L1_cache_size = atoi(GETENV("MR_L1CACHESIZE"));//1024 * 1024 * 2;
 map_reduce_args.num_map_threads = atoi(GETENV("MR_NUMTHREADS"));//8;
 map_reduce_args.num_reduce_threads = atoi(GETENV("MR_NUMTHREADS"));//16;
 map_reduce_args.num_merge_threads = atoi(GETENV("MR_NUMTHREADS"));//8;
 map_reduce_args.num_procs = atoi(GETENV("MR_NUMPROCS"));//16;
 map_reduce_args.key_match_factor = (float)atof(GETENV("MR_KEYMATCHFACTOR"));//2;
 printf("Wordcount: Calling MapReduce Scheduler Wordcount\n");
 gettimeofday(&starttime,0);
 get_time (&end);
#ifdef TIMING
 fprintf (stderr, "initialize: %u\n", time_diff (&end, &begin));
#endif
 get_time (&begin);
 CHECK_ERROR(map_reduce (&map_reduce_args) < 0);
 get_time (&end);
#ifdef TIMING
 library_time += time_diff (&end, &begin);
#endif
 get_time (&begin);
 gettimeofday(&endtime,0);
 printf("Wordcount: Completed %ld\n",(endtime.tv_sec - starttime.tv_sec));
 printf("Wordcount: MapReduce Completed\n");
 printf("Wordcount: Calling MapReduce Scheduler Sort\n");
 mapreduce_sort(wc_vals.data, wc_vals.length, sizeof(keyval_t), mykeyvalcmp);
 CHECK_ERROR (map_reduce_finalize ());
 printf("Wordcount: MapReduce Completed\n");
 dprintf("\nWordcount: Results (TOP %d):\n", disp_num);
 for (i = 0; i < disp_num && i < wc_vals.length; i++)
 {
 keyval_t * curr = &((keyval_t *)wc_vals.data)[i];
 dprintf("%15s - %" PRIdPTR "\n", (char *)curr->key, (intptr_t)curr->val);
 }
 free(wc_vals.data);
#ifndef NO_MMAP
 CHECK_ERROR(munmap(fdata, finfo.st_size + 1) < 0);
#else
 free (fdata);
#endif
 CHECK_ERROR(close(fd) < 0);
 get_time (&end);
#ifdef TIMING
 fprintf (stderr, "finalize: %u\n", time_diff (&end, &begin));
#endif
 return 0;
}

#ifndef __MAP_CU__
#define __MAP_CU__
#include "MarsInc.h"
#include "global.h"
__device__ int hash_func(char* str, int len)
{
int hash, i;
for (i = 0, hash=len; i < len; i++)
hash = (hash<<4)^(hash>>28)^str[i];
return hash;
}
__device__ void MAP_COUNT_FUNC//(void *key, void *val, size_t keySize, size_t valSize)
{
WC_KEY_T* pKey = (WC_KEY_T*)key;
WC_VAL_T* pVal = (WC_VAL_T*)val;
char* ptrBuf = pKey->file + pVal->line_offset;
int line_size = pVal->line_size;
char* p = ptrBuf;
int lsize = 0;
int wsize = 0;
char* start = ptrBuf;
while(1)
{
for (; *p >= 'A' && *p <= 'Z'; p++, lsize++);
*p = '\0';
++p;
++lsize;
wsize = (int)(p - start);
if (wsize > 6)
{
//printf("%s, wsize:%d\n", start, wsize);
EMIT_INTER_COUNT_FUNC(wsize, sizeof(int));
}
for (; (lsize < line_size) && (*p < 'A' || *p > 'Z'); p++, lsize++);
if (lsize >= line_size) break;
start = p;
}
}
__device__ void MAP_FUNC//(void *key, void val, size_t keySize, size_t valSize)
{
WC_KEY_T* pKey = (WC_KEY_T*)key;
WC_VAL_T* pVal = (WC_VAL_T*)val;
char* filebuf = pKey->file;
char* ptrBuf = filebuf + pVal->line_offset;
int line_size = pVal->line_size;
char* p = ptrBuf;
char* start = ptrBuf;
int lsize = 0;
int wsize = 0;
while(1)
{
for (; *p >= 'A' && *p <= 'Z'; p++, lsize++);
*p = '\0';
++p;
++lsize;
wsize = (int)(p - start);
int* o_val = (int*)GET_OUTPUT_BUF(0);
*o_val = wsize;
if (wsize > 6) 
{
//printf("%s, %d\n", start, wsize);
EMIT_INTERMEDIATE_FUNC(start, o_val, wsize, sizeof(int));
}
for (; (lsize < line_size) && (*p < 'A' || *p > 'Z'); p++, lsize++);
if (lsize >= line_size) break;
start = p;
}
}
#endif //__MAP_CU__

#ifndef __REDUCE_CU__
#define __REDUCE_CU__
#include "MarsInc.h"
__device__ void REDUCE_COUNT_FUNC//(void* key, void* vals, size_t keySize, size_t valCount)
{
}
__device__ void REDUCE_FUNC//(void* key, void* vals, size_t keySize, size_t valCount)
{
}
#endif //__REDUCE_CU__