使用Hadoop MapWritable数据类型
MapWritable：这是用来存储一个key-value对的map。key和value应该都是Writable数据类型。 可以像下面这样使用该MapWritable函数。
MapWritable valueMap = new MapWritable();
valueMap.put(new IntWritable(1),new Text("test"));
不过，我们应该意识到，由于包含了存储在该map中的每个对象的类名，所以MapWritable的序列化增加了轻微的性能损失。
案例描述
现在有如下的员工信息：
这些信息存储在数据文件input.txt中，如下：
1201,gopal,45,Male,50000 1202,manisha,40,Female,51000 1203,khaleel,34,Male,30000 1204,prasanth,30,Male,31000 1205,kiran,20,Male,40000 1206,laxmi,25,Female,35000 1207,bhavya,20,Female,15000 1208,reshma,19,Female,14000 1209,kranthi,22,Male,22000 1210,Satish,24,Male,25000 1211,Krishna,25,Male,26000 1212,Arshad,28,Male,20000 1213,lavanya,18,Female,8000
要求编写MapReduce应用程序，处理输入数据集，按性别找出不同年龄组中最高薪水的员工 (例如, 小于20岁, 21岁至30岁之间, 大于30岁)，并输出。
一、创建Java Maven项目
Maven依赖：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>HadoopDemo</groupId>
<artifactId>com.xueai8</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!--hadoop依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.1</version>
</dependency>
<!--hdfs文件系统依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>3.3.1</version>
</dependency>
<!--MapReduce相关的依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>3.3.1</version>
</dependency>
<!--junit依赖-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!--编译器插件用于编译拓扑-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source> <!-- 源代码使用的JDK版本 -->
<target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
<encoding>UTF-8</encoding><!-- 字符集编码 -->
</configuration>
</plugin>
</plugins>
</build>
</project>
MapWriteMapper.java：
package com.xueai8.mapwritable;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class MapWriteMapper extends Mapper<LongWritable, Text, Text, MapWritable> {
private final Text genderKey = new Text();
private final MapWritable mapValue = new MapWritable();
// 一行输入数据：1201,gopal,45,Male,50000
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] str = value.toString().split(",", -1); // 分词
int age = Integer.parseInt(str[2]); // 取年龄
int salary = Integer.parseInt(str[4]); // 取薪资
mapValue.put(new Text("age"), new IntWritable(age));
mapValue.put(new Text("salary"), new IntWritable(salary));
genderKey.set(str[3]);
context.write(genderKey, mapValue); // 写出
}
}
CaderPartitioner.java：自定义分区器
package com.xueai8.mapwritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
// 根据给定的分区条件，输入的key-value对数据基于年龄条件可以被分为三部分
public class CaderPartitioner extends Partitioner<Text, MapWritable> {
@Override
public int getPartition(Text key, MapWritable value, int numReduceTasks) {
// 从MapWritable中按key对值
Text ageKey = new Text("age");
IntWritable ageWritable = (IntWritable)value.get(ageKey);
// IntWritable ageWritable = (IntWritable)value.get(new Text("age"));
// 转换为java的int类型
int age = ageWritable.get();
// 计算并返回分区id
if (numReduceTasks == 0) {
return 0;
}
if (age <= 20) {
return 0;
} else if (age > 20 && age <= 30) {
return 1 % numReduceTasks;
} else {
return 2 % numReduceTasks;
}
}
}
MapWriteReducer.java：
package com.xueai8.mapwritable;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class MapWriteReducer extends Reducer<Text, MapWritable, Text, IntWritable> {
private final Text outKey = new Text();
private final IntWritable outValue = new IntWritable(0);
@Override
public void reduce(Text key, Iterable<MapWritable> values, Context context)
throws IOException, InterruptedException {
int max = -1;
// 遍历值迭代器，计算每个key(性别)的最大薪资
for (MapWritable val : values) {
IntWritable salary = (IntWritable) val.get(new Text("salary"));
if (salary.get() > max) {
max = salary.get();
}
}
outKey.set(key);
outValue.set(max);
context.write(outKey, outValue);
}
}
MapWriteDriver.java：
package com.xueai8.mapwritable;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class MapWriteDriver {
public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {
if (args.length < 2) {
System.err.println("用法: MapWriteDriver <input_path> <output_path>");
System.exit(-1);
}
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "PartitionerDemo");
job.setJarByClass(MapWriteDriver.class);
// set mapper
job.setMapperClass(MapWriteMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(MapWritable.class);
// 设置自定义分区器
job.setPartitionerClass(CaderPartitioner.class);
// set reducer
job.setReducerClass(MapWriteReducer.class);
job.setNumReduceTasks(3); // 指定三个分区
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
System.exit(job.waitForCompletion(true) ? 0 : -1);
}
}
二、配置log4j
在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：
log4j.rootLogger = info,stdout
### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
三、项目打包
打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：
如果一切正常，会提示打jar包成功。如下图所示：
这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：
四、项目部署
请按以下步骤执行。
1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：
$ start-dfs.sh $ start-yarn.sh
查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：
$ jps
这时应该能看到有如下5个进程正在运行，说明集群运行正常：
5542 NodeManager 5191 SecondaryNameNode 4857 NameNode 5418 ResourceManager 4975 DataNode
2、将数据文件sample.txt上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put input.txt /data/mr/ $ hdfs dfs -ls /data/mr/
3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）
在终端窗口中，执行如下的作业提交命令：
$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.mapwritable.MapWriteDriver /data/mr /data/mr-output
4、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output
会发现生成了三个输出结果文件，每个reducer(每个分区)对应一个输出文件。
查看第一个结果文件中的内容：
$ hdfs dfs -cat /data/mr-output/part-r-00000
可以看到如下的输出结果：
Female 15000 Male 40000
查看第二个结果文件中的内容：
$ hdfs dfs -cat /data/mr-output/part-r-00001
可以看到如下的输出结果：
Female 35000 Male 31000
查看第三个结果文件中的内容：
$ hdfs dfs -cat /data/mr-output/part-r-00002
可以看到如下的输出结果：
Female 51000 Male 50000
