学步园

今天从网上看到点数据，很适合用MapReduce来分析一下。一条记录的格式如下：

[**] [1:538:15] NETBIOS SMB IPC$ unicode share access [**]
[Classification: Generic Protocol Command Decode] [Priority: 3]
09/04-17:53:56.363811 168.150.177.165:1051 -> 168.150.177.166:139
TCP TTL:128 TOS:0x0 ID:4000 IpLen:20 DgmLen:138 DF
***AP*** Seq: 0x2E589B8  Ack: 0x642D47F9  Win: 0x4241  TcpLen: 20

[**] [1:1917:6] SCAN UPnP service discover attempt [**]
[Classification: Detection of a Network Scan] [Priority: 3]
09/04-17:53:56.385573 168.150.177.164:1032 -> 239.255.255.250:1900
UDP TTL:1 TOS:0x0 ID:80 IpLen:20 DgmLen:161
Len: 133

大家可以看到要处理上面的记录，肯定不能用系统默认的TextInputFormat.class

所以要自己写一个读取类，从上面的格式可以看出。每一条记录由换行符分割，其余的行为一条记录（包括多行）。闲话少说，直接上代码：

public class MyRecordReader extends RecordReader<IntWritable, Text>{<br /> private static final Log LOG = LogFactory.getLog(MyRecordReader.class);</p> <p> private int pos;<br /> private boolean more;<br /> private LineReader in;<br /> private int maxLineLength;<br /> private IntWritable key = null;<br /> private Text value1 = null;<br /> private String value = "";</p> <p> public void initialize(InputSplit genericSplit,<br /> TaskAttemptContext context) throws IOException {<br /> pos = 1;<br /> more = true;<br /> FileSplit split = (FileSplit) genericSplit;<br /> Configuration job = context.getConfiguration();<br /> this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",<br /> Integer.MAX_VALUE);</p> <p> final Path file = split.getPath();<br /> // open the file and seek to the start of the split<br /> FileSystem fs = file.getFileSystem(job);<br /> FSDataInputStream fileIn = fs.open(split.getPath());</p> <p> in = new LineReader(fileIn, job);<br /> }</p> <p> public boolean nextKeyValue() throws IOException {<br /> if (key == null) {<br /> key = new IntWritable();<br /> }<br /> key.set(pos);<br /> if (value1 == null) {<br /> value1 = new Text();<br /> }<br /> value = "";<br /> int newSize = 0;<br /> while (true) {<br /> newSize = in.readLine(value1, maxLineLength,maxLineLength);<br /> pos++;</p> <p> if (newSize == 0) {<br /> //value += value1.toString();<br /> if (!value.isEmpty()){<br /> newSize = 1;<br /> }<br /> break;<br /> }<br /> if (newSize == 1) {<br /> //当newSize == 1是就是读取的换行符，所以要输出。<br /> break;<br /> }<br /> if (newSize < maxLineLength) {<br /> //如果大于1，就证明读取了一行，但这条记录并没有结束。<br /> value += value1.toString();<br /> //中间用空格分割一下<br /> value += " ";<br /> //break;<br /> }<br /> // line too long. try again<br /> LOG.info("Skipped line of size " + newSize + " at pos " +<br /> (pos - newSize));<br /> }<br /> if (newSize == 0) {<br /> key = null;<br /> //value = "";<br /> more = false;<br /> return false;<br /> } else {<br /> return true;<br /> }<br /> }<br /> @Override<br /> public IntWritable getCurrentKey() {<br /> return key;<br /> }<br /> @Override<br /> public Text getCurrentValue() {<br /> return new Text(value);<br /> }<br /> /**<br /> * Get the progress within the split<br /> */<br /> public float getProgress() {<br /> if (more) {<br /> return 0.0f;<br /> } else {<br /> return 100;<br /> }<br /> }</p> <p> public synchronized void close() throws IOException {<br /> if (in != null) {<br /> in.close();<br /> }<br /> }<br /> }<br />

通过上面的类，就可以将4行连接为一条记录。换行符作为一条记录的结束。