MapReduce OutputFormat 高级定制：多输出与自定义分隔符，让数据“各回各家，各找各妈”！
各位观众老爷，各位编程英雄，欢迎来到今天的“数据炼金术”课堂！🧙♂️ 今天我们要聊点高阶的，让我们的 MapReduce 程序不仅能“吃得饱”（处理海量数据），还能“吃得好”（输出格式优雅）。
我们都知道，MapReduce 框架就像一个超级工厂，输入一堆原材料（输入数据），经过“Map”和“Reduce”两道大工序，最终产出成品（输出数据）。但是，默认情况下，这个工厂只会把所有成品一股脑儿地堆在一个仓库里（一个输出文件），而且成品之间还用默认的分隔符（通常是制表符和换行符）隔开，就像一锅乱炖，毫无美感。
如果你想让你的数据“各回各家，各找各妈”，实现多输出，或者想让你的数据排列得整整齐齐，使用自定义分隔符，那么你就需要深入了解 MapReduce 的 OutputFormat 了。
今天，我们就来一起探索 OutputFormat 的高级定制，让你的 MapReduce 程序输出的数据更加灵活、美观、实用！
1. OutputFormat：数据的“最终归宿”
首先，让我们来认识一下 OutputFormat。它就像 MapReduce 程序的“出口”，负责将 Reduce 阶段处理后的数据写入到存储系统中（例如 HDFS）。
OutputFormat 是一个抽象类，定义了数据输出的基本接口。Hadoop 已经为我们提供了一些常用的 OutputFormat 实现，例如：
TextOutputFormat: 这是最常用的OutputFormat，它以文本格式输出数据，Key 和 Value 之间用制表符分隔，每条记录之间用换行符分隔。SequenceFileOutputFormat: 以 SequenceFile 格式输出数据，这是一种 Hadoop 特有的二进制文件格式，适合存储大量小文件。MapFileOutputFormat: 以 MapFile 格式输出数据，MapFile 是一种排序后的 SequenceFile，可以根据 Key 进行快速查找。MultipleOutputs: 这不是一个真正的OutputFormat，而是一个工具类，可以帮助我们实现多输出。
但是，这些默认的 OutputFormat 往往不能满足我们所有的需求。例如，我们需要将不同类型的数据输出到不同的文件中，或者我们需要使用自定义的分隔符来分隔 Key 和 Value。这时，我们就需要自定义 OutputFormat 了。
2. 多输出：让数据“各得其所”
想象一下，你的 MapReduce 程序处理的是用户行为数据，你想要将不同类型的用户行为分别输出到不同的文件中，例如，将“点击”行为输出到 clicks.txt，将“浏览”行为输出到 views.txt，将“购买”行为输出到 purchases.txt。
这时，MultipleOutputs 就派上用场了！它允许我们将数据输出到多个不同的文件中，每个文件对应一种特定的数据类型。
如何使用 MultipleOutputs？
使用 MultipleOutputs 非常简单，只需要以下几个步骤：
-
在 Mapper 或 Reducer 中创建
MultipleOutputs对象：private MultipleOutputs<Text, Text> multipleOutputs; @Override protected void setup(Context context) throws IOException, InterruptedException { multipleOutputs = new MultipleOutputs<>(context); } -
在 Mapper 或 Reducer 中使用
write()方法输出数据到不同的文件中：@Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { for (Text value : values) { String behaviorType = value.toString().split(":")[0]; // 假设 value 的格式是 "behaviorType:data" switch (behaviorType) { case "click": multipleOutputs.write("clicks", key, value); // 输出到 clicks-r-xxxxx 文件 break; case "view": multipleOutputs.write("views", key, value); // 输出到 views-r-xxxxx 文件 break; case "purchase": multipleOutputs.write("purchases", key, value); // 输出到 purchases-r-xxxxx 文件 break; default: context.write(key, value); // 输出到默认的输出文件 } } }multipleOutputs.write(String namedOutput, KEY key, VALUE value): 将数据输出到名为namedOutput的文件中。namedOutput：是一个字符串，用于指定输出文件的名称前缀。例如，如果namedOutput是 "clicks"，那么输出文件的名称将会是clicks-r-xxxxx，其中xxxxx是 Reduce Task 的 ID。
-
在 Mapper 或 Reducer 的
cleanup()方法中关闭MultipleOutputs对象：@Override protected void cleanup(Context context) throws IOException, InterruptedException { multipleOutputs.close(); } -
在 Job 配置中配置
MultipleOutputs：Job job = Job.getInstance(conf, "Multiple Outputs Example"); // ... 其他配置 MultipleOutputs.addNamedOutput(job, "clicks", TextOutputFormat.class, Text.class, Text.class); MultipleOutputs.addNamedOutput(job, "views", TextOutputFormat.class, Text.class, Text.class); MultipleOutputs.addNamedOutput(job, "purchases", TextOutputFormat.class, Text.class, Text.class);MultipleOutputs.addNamedOutput(Job job, String namedOutput, Class<? extends OutputFormat<KEY, VALUE>> outputFormatClass, Class<?> keyClass, Class<?> valueClass): 配置一个名为namedOutput的输出，指定其OutputFormat类，Key 类和 Value 类。
示例代码：
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;
public class MultipleOutputsExample {
public static class MyMapper extends Mapper<Object, Text, Text, Text> {
private Text user = new Text();
private Text behavior = new Text();
@Override
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String[] parts = value.toString().split(","); // 假设输入数据格式是 "user,behaviorType:data"
user.set(parts[0]);
behavior.set(parts[1]);
context.write(user, behavior);
}
}
public static class MyReducer extends Reducer<Text, Text, Text, Text> {
private MultipleOutputs<Text, Text> multipleOutputs;
@Override
protected void setup(Context context) throws IOException, InterruptedException {
multipleOutputs = new MultipleOutputs<>(context);
}
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
for (Text value : values) {
String behaviorType = value.toString().split(":")[0]; // 假设 value 的格式是 "behaviorType:data"
switch (behaviorType) {
case "click":
multipleOutputs.write("clicks", key, value); // 输出到 clicks-r-xxxxx 文件
break;
case "view":
multipleOutputs.write("views", key, value); // 输出到 views-r-xxxxx 文件
break;
case "purchase":
multipleOutputs.write("purchases", key, value); // 输出到 purchases-r-xxxxx 文件
break;
default:
context.write(key, value); // 输出到默认的输出文件
}
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
multipleOutputs.close();
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Multiple Outputs Example");
job.setJarByClass(MultipleOutputsExample.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
MultipleOutputs.addNamedOutput(job, "clicks", TextOutputFormat.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(job, "views", TextOutputFormat.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(job, "purchases", TextOutputFormat.class, Text.class, Text.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}总结：
MultipleOutputs 就像一个“数据分拣员”，它可以根据数据的类型，将数据分发到不同的文件中，让你的数据“各得其所”，方便后续的分析和处理。
3. 自定义分隔符：让数据“井然有序”
默认情况下，TextOutputFormat 使用制表符来分隔 Key 和 Value，使用换行符来分隔每条记录。但是，有时候我们需要使用自定义的分隔符，例如，使用逗号分隔 Key 和 Value，或者使用自定义的字符串来分隔每条记录。
这时，我们就需要自定义 OutputFormat 了。
如何自定义 OutputFormat？
自定义 OutputFormat 需要实现以下几个步骤：
-
创建一个类，继承
TextOutputFormat：public class MyCustomOutputFormat extends TextOutputFormat<Text, Text> { // ... } -
重写
getRecordWriter()方法，返回一个自定义的RecordWriter：@Override public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException { Configuration conf = job.getConfiguration(); String keyValueSeparator = conf.get("mapreduce.output.key.value.separator", "t"); // 从配置中获取分隔符 Path file = getDefaultWorkFile(job, ""); FileSystem fs = file.getFileSystem(conf); FSDataOutputStream fileOut = fs.create(file, false); return new MyCustomRecordWriter(fileOut, keyValueSeparator); }getRecordWriter()方法负责创建一个RecordWriter对象，该对象负责将数据写入到输出文件中。TaskAttemptContext：包含 Task 的上下文信息，例如 Job 配置。getDefaultWorkFile()：获取默认的工作文件路径。FileSystem：Hadoop 文件系统对象。FSDataOutputStream：文件输出流。
-
创建一个类，继承
RecordWriter：public static class MyCustomRecordWriter extends RecordWriter<Text, Text> { private static final String utf8 = "UTF-8"; private FSDataOutputStream out; private final byte[] keyValueSeparator; public MyCustomRecordWriter(FSDataOutputStream out, String keyValueSeparator) { this.out = out; try { this.keyValueSeparator = keyValueSeparator.getBytes(utf8); } catch (UnsupportedEncodingException e) { throw new IllegalArgumentException("can't find " + utf8 + " encoding"); } } @Override public void write(Text key, Text value) throws IOException, InterruptedException { if (key != null) { out.write(key.getBytes(), 0, key.getLength()); out.write(keyValueSeparator); } if (value != null) { out.write(value.getBytes(), 0, value.getLength()); } out.write("n".getBytes(), 0, "n".getBytes().length); // 添加换行符 } @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { out.close(); } }RecordWriter负责将 Key 和 Value 写入到输出流中。write()方法：将 Key 和 Value 写入到输出流中，并使用自定义的分隔符分隔。close()方法：关闭输出流。
-
在 Job 配置中配置自定义的
OutputFormat：Job job = Job.getInstance(conf, "Custom Output Format Example"); // ... 其他配置 job.setOutputFormatClass(MyCustomOutputFormat.class); conf.set("mapreduce.output.key.value.separator", ","); // 设置 Key-Value 分隔符为逗号job.setOutputFormatClass(MyCustomOutputFormat.class): 设置自定义的OutputFormat类。conf.set("mapreduce.output.key.value.separator", ","): 设置 Key-Value 分隔符为逗号。
示例代码：
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
public class CustomOutputFormatExample {
public static class MyCustomOutputFormat extends TextOutputFormat<Text, Text> {
@Override
public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
Configuration conf = job.getConfiguration();
String keyValueSeparator = conf.get("mapreduce.output.key.value.separator", "t"); // 从配置中获取分隔符
Path file = getDefaultWorkFile(job, "");
FileSystem fs = file.getFileSystem(conf);
FSDataOutputStream fileOut = fs.create(file, false);
return new MyCustomRecordWriter(fileOut, keyValueSeparator);
}
}
public static class MyCustomRecordWriter extends RecordWriter<Text, Text> {
private static final String utf8 = "UTF-8";
private FSDataOutputStream out;
private final byte[] keyValueSeparator;
public MyCustomRecordWriter(FSDataOutputStream out, String keyValueSeparator) {
this.out = out;
try {
this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("can't find " + utf8 + " encoding");
}
}
@Override
public void write(Text key, Text value) throws IOException, InterruptedException {
if (key != null) {
out.write(key.getBytes(), 0, key.getLength());
out.write(keyValueSeparator);
}
if (value != null) {
out.write(value.getBytes(), 0, value.getLength());
}
out.write("n".getBytes(), 0, "n".getBytes().length); // 添加换行符
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
out.close();
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Custom Output Format Example");
job.setJarByClass(CustomOutputFormatExample.class);
// ... 其他配置 (Mapper, Reducer, InputFormat, etc.)
job.setOutputFormatClass(MyCustomOutputFormat.class);
conf.set("mapreduce.output.key.value.separator", ","); // 设置 Key-Value 分隔符为逗号
// 模拟数据输入
// ...
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}总结：
自定义 OutputFormat 就像一个“数据格式化器”，它可以根据你的需求，将数据格式化成你想要的格式，让你的数据“井然有序”，方便后续的读取和使用。
4. 案例分析：电商数据分析
让我们来看一个实际的案例，假设我们正在进行电商数据分析，需要分析用户的购买行为。我们有以下需求：
- 将不同品类的商品购买数据输出到不同的文件中，例如，将“服装”品类的购买数据输出到
clothing.txt，将“电子产品”品类的购买数据输出到electronics.txt。 - 使用逗号分隔用户 ID 和购买金额。
我们可以结合 MultipleOutputs 和自定义 OutputFormat 来实现这个需求。
- 使用
MultipleOutputs将不同品类的商品购买数据输出到不同的文件中。 - 自定义
OutputFormat，使用逗号分隔用户 ID 和购买金额。
实现步骤：
-
创建自定义
OutputFormat：public class CommaSeparatedOutputFormat extends TextOutputFormat<Text, Text> { @Override public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException { Configuration conf = job.getConfiguration(); Path file = getDefaultWorkFile(job, ""); FileSystem fs = file.getFileSystem(conf); FSDataOutputStream fileOut = fs.create(file, false); return new CommaSeparatedRecordWriter(fileOut); } } public static class CommaSeparatedRecordWriter extends RecordWriter<Text, Text> { private static final String utf8 = "UTF-8"; private FSDataOutputStream out; public CommaSeparatedRecordWriter(FSDataOutputStream out) { this.out = out; } @Override public void write(Text key, Text value) throws IOException, InterruptedException { if (key != null) { out.write(key.getBytes(), 0, key.getLength()); out.write(",".getBytes(utf8)); // 使用逗号分隔 } if (value != null) { out.write(value.getBytes(), 0, value.getLength()); } out.write("n".getBytes(utf8), 0, "n".getBytes(utf8).length); } @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { out.close(); } } -
在 Reducer 中使用
MultipleOutputs和自定义OutputFormat：public static class MyReducer extends Reducer<Text, Text, Text, Text> { private MultipleOutputs<Text, Text> multipleOutputs; @Override protected void setup(Context context) throws IOException, InterruptedException { multipleOutputs = new MultipleOutputs<>(context); } @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { for (Text value : values) { String category = value.toString().split(":")[0]; // 假设 value 的格式是 "category:amount" switch (category) { case "clothing": multipleOutputs.write("clothing", key, value, CommaSeparatedOutputFormat.class); // 输出到 clothing-r-xxxxx 文件，使用逗号分隔 break; case "electronics": multipleOutputs.write("electronics", key, value, CommaSeparatedOutputFormat.class); // 输出到 electronics-r-xxxxx 文件，使用逗号分隔 break; default: context.write(key, value); // 输出到默认的输出文件 } } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { multipleOutputs.close(); } } -
在 Job 配置中配置
MultipleOutputs和自定义OutputFormat：Job job = Job.getInstance(conf, "Ecommerce Data Analysis"); // ... 其他配置 MultipleOutputs.addNamedOutput(job, "clothing", CommaSeparatedOutputFormat.class, Text.class, Text.class); MultipleOutputs.addNamedOutput(job, "electronics", CommaSeparatedOutputFormat.class, Text.class, Text.class);
通过以上步骤，我们就可以实现将不同品类的商品购买数据输出到不同的文件中，并使用逗号分隔用户 ID 和购买金额。
5. 总结
今天，我们深入学习了 MapReduce OutputFormat 的高级定制，包括多输出和自定义分隔符。
MultipleOutputs可以帮助我们将数据输出到多个不同的文件中，让数据“各得其所”。- 自定义
OutputFormat可以帮助我们将数据格式化成我们想要的格式，让数据“井然有序”。
掌握了这些技巧，你就可以让你的 MapReduce 程序输出的数据更加灵活、美观、实用，从而更好地服务于你的数据分析任务。
希望今天的课程对你有所帮助！下次再见！👋