在上一篇中，我们实现了按 cookieId 和 time 进行二次排序，现在又有新问题：假如我需要按 cookieId 和 cookieId&time 的组合进行分析呢？此时最好的办法是自定义 InputFormat，让 mapreduce 一次读取一个 cookieId 下的所有记录，然后再按 time 进行切分 session，逻辑伪码如下：
for OneSplit in MyInputFormat.getSplit() // OneSplit 是某个 cookieId 下的所有记录
for session in OneSplit // session 是按 time 把 OneSplit 进行了二次分割
for line in session // line 是 session 中的每条记录，对应原始日志的某条记录
1、原理：
InputFormat是MapReduce中一个很常用的概念，它在程序的运行中到底起到了什么作用呢？
InputFormat其实是一个接口，包含了两个方法：
public interface InputFormat<K, V> {
InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> createRecordReader(InputSplit split,
TaskAttemptContext context) throws IOException;
}
这两个方法有分别完成着以下工作：
方法 getSplits 将输入数据切分成splits，splits的个数即为map tasks的个数，splits的大小默认为块大小，即64M
方法 getRecordReader 将每个 split 解析成records, 再依次将record解析成<K,V>对
也就是说 InputFormat完成以下工作：
InputFile --> splits --> <K,V>
系统常用的 InputFormat 又有哪些呢？
其中Text InputFormat便是最常用的，它的 <K,V>就代表 <行偏移,该行内容>
然而系统所提供的这几种固定的将 InputFile转换为 <K,V>的方式有时候并不能满足我们的需求：
此时需要我们自定义 InputFormat ，从而使Hadoop框架按照我们预设的方式来将
InputFile解析为<K,V>
在领会自定义 InputFormat 之前，需要弄懂一下几个抽象类、接口及其之间的关系：
InputFormat(interface), FileInputFormat(abstract class), TextInputFormat(class),
RecordReader (interface), Line RecordReader(class)的关系
FileInputFormat implements InputFormat
TextInputFormat extends FileInputFormat
TextInputFormat.get RecordReader calls Line RecordReader
Line RecordReader implements RecordReader
对于InputFormat接口，上面已经有详细的描述
再看看 FileInputFormat，它实现了 InputFormat接口中的 getSplits方法，而将 getRecordReader与isSplitable留给具体类(如 TextInputFormat )实现， isSplitable方法通常不用修改，所以只需要在自定义的 InputFormat中实现
getRecordReader方法即可，而该方法的核心是调用 Line RecordReader(即由LineRecorderReader类来实现 " 将每个s plit解析成records, 再依次将record解析成<K,V>对" )，该方法实现了接口RecordReader
public interface RecordReader<K, V> {
boolean next(K key, V value) throws IOException; K createKey();
V createValue();
long getPos() throws IOException;
public void close() throws IOException;
float getProgress() throws IOException;
}
因此自定义InputFormat的核心是自定义一个实现接口RecordReader类似于LineRecordReader的类，该类的核心也正是重写接口RecordReader中的几大方法，
定义一个InputFormat的核心是定义一个类似于LineRecordReader的，自己的RecordReader
2、代码：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
package MyInputFormat;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
public class TrackInputFormat extends FileInputFormat<LongWritable, Text> {
@SuppressWarnings("deprecation")
@Override
public RecordReader<LongWritable, Text> createRecordReader(
InputSplit split, TaskAttemptContext context) {
return new TrackRecordReader();
}
@Override
protected boolean isSplitable(JobContext context, Path file) {
CompressionCodec codec = new CompressionCodecFactory(
context.getConfiguration()).getCodec(file);
return codec == null;
}
} |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
package MyInputFormat;
import java.io.IOException;
import java.io.InputStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
/** * Treats keys as offset in file and value as line.
*
* @deprecated Use
* {@link org.apache.hadoop.mapreduce.lib.input.LineRecordReader}
* instead.
*/
public class TrackRecordReader extends RecordReader<LongWritable, Text> {
private static final Log LOG = LogFactory.getLog(TrackRecordReader.class);
private CompressionCodecFactory compressionCodecs = null;
private long start;
private long pos;
private long end;
private NewLineReader in;
private int maxLineLength;
private LongWritable key = null;
private Text value = null;
// ----------------------
// 行分隔符，即一条记录的分隔符
private byte[] separator = "END\n".getBytes();
// --------------------
public void initialize(InputSplit genericSplit, TaskAttemptContext context)
throws IOException {
FileSplit split = (FileSplit) genericSplit;
Configuration job = context.getConfiguration();
this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",
Integer.MAX_VALUE);
start = split.getStart();
end = start + split.getLength();
final Path file = split.getPath();
compressionCodecs = new CompressionCodecFactory(job);
final CompressionCodec codec = compressionCodecs.getCodec(file);
FileSystem fs = file.getFileSystem(job);
FSDataInputStream fileIn = fs.open(split.getPath());
boolean skipFirstLine = false;
if (codec != null) {
in = new NewLineReader(codec.createInputStream(fileIn), job);
end = Long.MAX_VALUE;
} else {
if (start != 0) {
skipFirstLine = true;
this.start -= separator.length;//
// --start;
fileIn.seek(start);
}
in = new NewLineReader(fileIn, job);
}
if (skipFirstLine) { // skip first line and re-establish "start".
start += in.readLine(new Text(), 0,
(int) Math.min((long) Integer.MAX_VALUE, end - start));
}
this.pos = start;
}
public boolean nextKeyValue() throws IOException {
if (key == null) {
key = new LongWritable();
}
key.set(pos);
if (value == null) {
value = new Text();
}
int newSize = 0;
while (pos < end) {
newSize = in.readLine(value, maxLineLength,
Math.max((int) Math.min(Integer.MAX_VALUE, end - pos),
maxLineLength));
if (newSize == 0) {
break;
}
pos += newSize;
if (newSize < maxLineLength) {
break;
}
LOG.info("Skipped line of size " + newSize + " at pos "
+ (pos - newSize));
}
if (newSize == 0) {
key = null;
value = null;
return false;
} else {
return true;
}
}
@Override
public LongWritable getCurrentKey() {
return key;
}
@Override
public Text getCurrentValue() {
return value;
}
/**
* Get the progress within the split
*/
public float getProgress() {
if (start == end) {
return 0.0f;
} else {
return Math.min(1.0f, (pos - start) / (float) (end - start));
}
}
public synchronized void close() throws IOException {
if (in != null) {
in.close();
}
}
public class NewLineReader {
private static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
private int bufferSize = DEFAULT_BUFFER_SIZE;
private InputStream in;
private byte[] buffer;
private int bufferLength = 0;
private int bufferPosn = 0;
public NewLineReader(InputStream in) {
this(in, DEFAULT_BUFFER_SIZE);
}
public NewLineReader(InputStream in, int bufferSize) {
this.in = in;
this.bufferSize = bufferSize;
this.buffer = new byte[this.bufferSize];
}
public NewLineReader(InputStream in, Configuration conf)
throws IOException {
this(in, conf.getInt("io.file.buffer.size", DEFAULT_BUFFER_SIZE));
}
public void close() throws IOException {
in.close();
}
public int readLine(Text str, int maxLineLength, int maxBytesToConsume)
throws IOException {
str.clear();
Text record = new Text();
int txtLength = 0;
long bytesConsumed = 0L;
boolean newline = false;
int sepPosn = 0;
do {
// 已经读到buffer的末尾了，读下一个buffer
if (this.bufferPosn >= this.bufferLength) {
bufferPosn = 0;
bufferLength = in.read(buffer);
// 读到文件末尾了，则跳出，进行下一个文件的读取
if (bufferLength <= 0) {
break;
}
}
int startPosn = this.bufferPosn;
for (; bufferPosn < bufferLength; bufferPosn++) {
// 处理上一个buffer的尾巴被切成了两半的分隔符(如果分隔符中重复字符过多在这里会有问题)
if (sepPosn > 0 && buffer[bufferPosn] != separator[sepPosn]) {
sepPosn = 0;
}
// 遇到行分隔符的第一个字符
if (buffer[bufferPosn] == separator[sepPosn]) {
bufferPosn++;
int i = 0;
// 判断接下来的字符是否也是行分隔符中的字符
for (++sepPosn; sepPosn < separator.length; i++, sepPosn++) {
// buffer的最后刚好是分隔符，且分隔符被不幸地切成了两半
if (bufferPosn + i >= bufferLength) {
bufferPosn += i - 1;
break;
}
// 一旦其中有一个字符不相同，就判定为不是分隔符
if (this.buffer[this.bufferPosn + i] != separator[sepPosn]) {
sepPosn = 0;
break;
}
}
// 的确遇到了行分隔符
if (sepPosn == separator.length) {
bufferPosn += i;
newline = true;
sepPosn = 0;
break;
}
}
}
int readLength = this.bufferPosn - startPosn;
bytesConsumed += readLength;
// 行分隔符不放入块中
if (readLength > maxLineLength - txtLength) {
readLength = maxLineLength - txtLength;
}
if (readLength > 0) {
record.append(this.buffer, startPosn, readLength);
txtLength += readLength;
// 去掉记录的分隔符
if (newline) {
str.set(record.getBytes(), 0, record.getLength()
- separator.length);
}
}
} while (!newline && (bytesConsumed < maxBytesToConsume));
if (bytesConsumed > (long) Integer.MAX_VALUE) {
throw new IOException("Too many bytes before newline: "
+ bytesConsumed);
}
return (int) bytesConsumed;
}
public int readLine(Text str, int maxLineLength) throws IOException {
return readLine(str, maxLineLength, Integer.MAX_VALUE);
}
public int readLine(Text str) throws IOException {
return readLine(str, Integer.MAX_VALUE, Integer.MAX_VALUE);
}
}
} |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
package MyInputFormat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
public class TestMyInputFormat {
public static class MapperClass extends Mapper<LongWritable, Text, Text, Text> {
public void map(LongWritable key, Text value, Context context) throws IOException,
InterruptedException {
System.out.println("key:\t " + key);
System.out.println("value:\t " + value);
System.out.println("-------------------------");
}
}
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
Path outPath = new Path("/hive/11");
FileSystem.get(conf).delete(outPath, true);
Job job = new Job(conf, "TestMyInputFormat");
job.setInputFormatClass(TrackInputFormat.class);
job.setJarByClass(TestMyInputFormat.class);
job.setMapperClass(TestMyInputFormat.MapperClass.class);
job.setNumReduceTasks(0);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.setOutputPath(job, outPath);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
} |
3、测试数据：
cookieId time url cookieOverFlag
|
1
2
3
4
5
6
7
8
9
|
1 a 1_hao123
1 a 1_baidu
1 b 1_google 2END
2 c 2_google
2 c 2_hao123
2 c 2_google 1END
3 a 3_baidu
3 a 3_sougou
3 b 3_soso 2END
|
4、结果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
key: 0
value: 1 a 1_hao123
1 a 1_baidu
1 b 1_google 2
-------------------------key: 47
value: 2 c 2_google
2 c 2_hao123
2 c 2_google 1
-------------------------key: 96
value: 3 a 3_baidu
3 a 3_sougou
3 b 3_soso 2
------------------------- |
REF：
自定义hadoop map/reduce输入文件切割InputFormat
http://hi.baidu.com/lzpsky/item/0d9d84c05afb43ba0c0a7b27
MapReduce高级编程之自定义InputFormat
http://datamining.xmu.edu.cn/bbs/home.php?mod=space&uid=91&do=blog&id=190
http://irwenqiang.iteye.com/blog/1448164
http://my.oschina.net/leejun2005/blog/133424
相关推荐
在Hadoop的生态系统中，MapReduce是处理海量数据的一种编程模型，而InputFormat作为MapReduce编程模型的重要组成部分，是负责处理输入数据的关键接口。为了深入理解MapReduce工作原理，必须掌握InputFormat的设计和...
1. Hadoop的MapReduce编程模型，包括InputFormat、Mapper、Partitioner、Reducer和OutputFormat等组件的作用。 2. Java的IO流和序列化，因为数据需要在网络间传输和持久化。 3. 分布式计算中的数据分区和排序，这...
2. InputFormat 模块负责做 Map 前的预处理，主要包括验证输入的格式是否符合 JobConfig 的输入定义、将 input 的文件切分为逻辑上的输入 InputSplit。 3. 将 RecordReader 处理后的结果作为 Map 的输入，然后 Map ...
在CountWord的例子中，Map阶段的主要任务是对输入的文本文件进行切分，通常是以行为单位。每行文本被分割成单词，并与一个计数值1关联。这个过程产生了许多键值对，键是单词，值是1。例如，输入文本"Hello World ...
4. **Hadoop API**：学习使用Hadoop API进行数据读写和处理，例如FileSystem API用于文件操作，InputFormat和OutputFormat定义输入输出格式，Mapper和Reducer实现数据处理逻辑。 5. **MapReduce编程**：理解...
Mapper接收InputFormat切分的数据，进行预处理，然后产生键值对；Reducer则接收Mapper的输出，进行聚合或汇总。用户可以自定义这两个阶段的行为，以适应各种计算需求。 4. **Partitioner**：Partitioner负责将...
9. **数据输入与输出**：Hadoop支持多种数据输入和输出格式，如TextInputFormat和TextInputFormat，以及自定义的InputFormat和OutputFormat，允许开发者处理不同类型的文件。 10. **Hadoop应用开发**：学习Hadoop...
Hadoop提供丰富的编程接口，如InputFormat和OutputFormat，RecordReader和RecordWriter等，源代码可以帮助理解如何自定义这些接口以适应不同数据格式和计算需求。 9. **Hadoop配置** 源代码中包含配置文件的解析...
Writable 是 Hadoop 自定义的序列化接口，实现该类的接口可以用作 MapReduce 过程中的 value 数据使用。 MapReduce 是一种高效的计算模型，通过将数据交给不同的机器去处理，数据划分，结果归约，实现了并行计算，...
1. **Input Splitting**：输入数据的切分仍由Java的InputFormat组件完成，它根据输入数据的特性将其分成若干个小块，这些小块将被分配给不同的Map任务进行处理。 2. **Map Task Execution**：每个Map任务由...
- **知识点说明**：Nagios本身是一款通用的系统监控工具，尽管它没有专门针对Hadoop的特性设计，但可以通过插件或自定义脚本来监控Hadoop集群的状态。 4. **如果NameNode意外终止，SecondaryNameNode会接替它使...
Map阶段的源码中，我们能看到InputFormat类负责切分输入数据，生成RecordReader对象，该对象读取数据并转化为键值对。Mapper类处理这些键值对，生成新的中间键值对。Map任务的输出会被分区和排序，然后通过网络传输...
在MapReduce编程模型中，map阶段的任务是处理输入数据，并生成键值对（key-value pairs），这些键值对会被传递给reduce阶段进行处理。reduce阶段的任务则通常是合并具有相同键的值。 理解了MapReduce的运行过程和...
默认的InputFormat是`TextInputFormat`，它的`getSplits()`方法会根据文件块大小和设定规则来决定切片，如切分规则为小于1.1倍大小的块不切分，大于0.1*block的会合并。 3. MapTask过程： MapTask执行流程分为五个...
Map 阶段的任务是对输入数据进行切分，然后对每一份数据进行处理，将原始文本分割成单词，并为每个单词生成键值对（<单词，1>）。Reduce 阶段则负责聚合 Map 阶段产生的结果，对相同单词的计数值进行累加。 ...
在IT行业中，MapReduce是一种分布式计算模型，常用于大数据处理，尤其在Apache Hadoop框架下。这个名为"my-test.zip"的压缩包文件似乎包含了使用MapReduce实现的三个具体统计案例，分别是统计流量、统计单词和统计...
例如，在Hadoop MapReduce中，InputFormat类负责将输入数据拆分为多个Map任务，每个任务处理一部分键值对。 接下来，List拆分与Map拆分类似，但更专注于有序或无序元素的序列。List通常包含一组元素，这些元素可以...
1. 面向接口编程：Hadoop大量使用Java的接口设计，如InputFormat、OutputFormat、Mapper和Reducer，允许开发者灵活地定义输入输出格式和处理逻辑。 2. 并发与网络编程：Java的并发库和Socket通信API在Hadoop中扮演...
- Map阶段：Map任务接收输入数据，将其切分为多个Split，由InputFormat的RecordReader将数据读取并转化为键值对（通常是, Text>）。Map函数处理这些键值对，产生新的中间键值对。默认情况下，Map的输出会被分区并...
2. **HDFS**：Hadoop分布式文件系统是一个高度容错性的系统，能够将数据分布在大量廉价硬件上，确保高可用性和数据可靠性。HDFS的设计原则是“一次写入，多次读取”。 3. **MapReduce**：MapReduce是Hadoop中的计算...