给我找一个用java实现hadoop的mapreduce编程的教程文档

请参考以下链接，里面详细介绍了如何使用Java实现Hadoop的MapReduce编程：

https://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html

相关问题

Hadoop_MapReduce编程实现 起码五个需求

### 使用Hadoop MapReduce编程实现五个及以上需求的示例

#### 1. 单词计数 (Word Count
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();

@Override
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
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();

@Override
protected 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);
}
}
}

此代码展示了如何创建一个简单的映射器和化简器来计算文档中各个词语的数量[^3]。

#### 2. 平均值计算 (Average Calculation)

平均值计算可以用来求解一组数值数据集的算术平均数。这通常涉及到两个阶段：先汇总总数与项数，再除以总数量得到最终结果。

// 假设有一个自定义的Combiner用于局部聚合...
public static class AvgMapper extends Mapper<LongWritable, Text, Text, DoubleArrayWritable> { ... }
public static class AvgReducer extends Reducer<Text, DoubleArrayWritable, Text, FloatWritable> { ... }

这里`DoubleArrayWritable`被设计成携带一对双精度浮点数——分别是累加后的总和及其对应的条目数目。

#### 3. 数据过滤 (Data Filtering)

通过设定特定条件筛选符合条件的数据记录。比如只保留年龄大于等于某个阈值的人的信息：

public static class FilterMapper extends Mapper<LongWritable, Text, NullWritable, Text> {

private final static NullWritable NULL_KEY = NullWritable.get();
private Text lineRecord;

@Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
this.lineRecord = new Text();
}

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] fields = value.toString().split(",");
if(Integer.parseInt(fields[1]) >= THRESHOLD_AGE){
lineRecord.set(value);
context.write(NULL_KEY,lineRecord );
}
}
}

// 这里不需要reducer因为只需要输出满足条件的结果即可。

这段代码实现了基于给定字段（如年龄）进行简单过滤的功能。

#### 4. 排序操作 (Sorting Operation)

对于大规模数据集来说，在分布式环境中执行排序是一项重要任务。可以通过设置适当的键比较逻辑来自定义排序顺序。

public static class SortMapper extends Mapper<LongWritable, Text, CustomKey, NullWritable> {...}
public static class SortReducer extends IdentityReducer<CustomKey,NullWritable>{...}

其中`CustomKey`应该继承于`WritableComparable<T>`接口并重写其compareTo方法以便支持自定义排序规则。

#### 5. 统计分析 (Statistical Analysis)

利用MapReduce框架来进行一些基本的概率分布估计或其他形式的统计学研究也是可行的；例如估算泊松分布参数λ或构建直方图等。

public static class StatisticMapper extends Mapper<LongWritable, Text, Text, LongWritable> {...}
public static class StatisticReducer extends Reducer<Text, LongWritable, Text, LongWritable> {...}

这些组件可以根据具体应用场景调整内部算法细节以适应不同的统计模型。

MapReduce编程实现单词统计

### 使用MapReduce编程实现单词统计

#### 项目配置
为了能够顺利开发并运行基于Hadoop的MapReduce应用，在项目的`pom.xml`文件中需引入必要的依赖项，这包括Hadoop客户端以及JUnit用于测试[^3]。

对于Maven管理的Java项目而言，具体的依赖声明如下所示：

<dependencies>
    <!-- hadoop客户端 -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.3.4</version>
    </dependency>
    <!-- 单元测试框架 -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.13.2</version>
    </dependency>
</dependencies>

#### 编写Mapper类
在WordCount案例中的映射器负责解析输入文本行，并输出每一对键值组合，其中键为单个词语而值则固定设为整数1表示该词出现了一次。此过程通过继承自`Mapper<LongWritable, Text, Text, IntWritable>`抽象基类的新子类完成定义[^1]。

public class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{
    
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer itr = new StringTokenizer(line);
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}

#### 编写Reducer类
归约者接收来自多个映射者的中间数据流作为输入参数；它按照相同的key聚合所有的value列表，之后计算这些数值之和得到最终的结果记录。这里采用的是累加操作来统计各个词条在整个文档集合里的总次数。

public class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {

    private IntWritable result = new IntWritable();

    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);
    }
}

#### 配置Job驱动程序
最后一步是在主函数内创建作业对象(Job)，设置好源路径(InputPath)与目标存储位置(OutputPath)，指定所使用的Mapper/Combiner/Reducer类型以及其他必要属性后即可调用`job.waitForCompletion(true)`方法触发整个流程执行[^2]。

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        
        job.setJarByClass(WordCountDriver.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(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}