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;
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();
public 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); // 输出<单词,1>
}
}
}
}
环境配置与项目搭建
Hadoop 3.x+Java 8开发环境
- 安装Java环境并配置
JAVA_HOME - 下载Hadoop二进制包解压,配置核心文件：
<!-- core-site.xml --> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property>
- Maven依赖配置：
<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.4</version> </dependency>
MapReduce核心编程模型
Mapper实现要点
- 继承
Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>基类 - 重写
map()方法处理输入拆分 - 使用
Context对象提交键值对
Reducer数据处理逻辑
public static 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); // 输出<单词,总次数>
}
}
作业调度与执行控制
驱动类配置最佳实践
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(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(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
性能优化关键策略
-
Combiner应用场景
- 在Mapper端预聚合数据
- 减少跨节点传输数据量
- 需满足交换律和结合律
-
分区优化技巧
job.setPartitionerClass(CustomPartitioner.class); // 自定义分区 job.setNumReduceTasks(4); // 匹配分区数
-
数据压缩配置
conf.set("mapreduce.map.output.compress", "true"); conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");
调试与异常处理
典型问题解决方案
- Task超时：调整配置参数
<property> <name>mapreduce.task.timeout</name> <value>600000</value> </property>
- 数据倾斜：实现自定义分区算法
- 内存溢出：优化JVM参数
conf.set("mapreduce.map.java.opts", "-Xmx2048m"); conf.set("mapreduce.reduce.java.opts", "-Xmx4096m");
生态整合应用
Hive集成方案
CREATE TABLE word_counts (word STRING, count INT) STORED AS ORC LOCATION '/output/wordcount';
Spark混合计算架构
val hadoopRDD = sc.newAPIHadoopFile(path,
classOf[TextInputFormat],
classOf[LongWritable],
classOf[Text])
您在实际开发中遇到哪些Hadoop性能瓶颈？是shuffle阶段数据交换效率问题，还是资源调度方面的挑战？欢迎分享具体场景，我们将针对性分析优化方案，对于千亿级数据集处理，您更倾向选择MapReduce还是Spark引擎？为什么？
原创文章，作者：世雄 - 原生数据库架构专家，如若转载，请注明出处：https://idctop.com/article/31140.html
