使用hadoop实现IP个数统计~并将结果写入数据库

浏览数：30 / 时间：2015年06月12日

转载请注明出处：http://blog.csdn.net/xiaojimanman/article/details/40372189

hadoop源代码中的WordCount事例中实现了单词统计，但是输出到HDFS文件，线上程序想使用其计算结果还要再次写个程序，所以自己就研究一下关于MapReduce的输出问题，下面就通过一个简单的例子说明下如何将MapReduce的计算结果输出到数据库中。

需求描述：

分析网络服务器上的Apache日志，统计每个IP访问资源的次数，并将结果写入到mysql数据库中。

数据格式：

Apache日志数据如下图所示：

一行数据就是一条http请求记录，该事例只做简单的IP个数统计。

需求分析：

通过MapReduce对日志文件采用分布式计算，map主要对日志做简单的拆分计数，reduce对map的结果求和。

map程序对一行日志数据做简单的拆分，获取客户端IP，输出结果为 key为客户端IP，value为IP出现次数。结果样例如下图所示：

reduce程序对Key值下的values做求和计算，输出结果为 key为客户端IP，value为IP出现次数。结果样例如下图所示：

上面的MapReduce程序和WordCount程序类似，只是对IP做了简单的求和计算，下面就需要写reduce的输出格式，使计算结果写入到mysql数据库中。

MapReduce支持用户自定义的输出格式，定义的类只需要继承FileOutputFormat即可。实现如下图所示：

自定义输出需要实现getRecordWriter方法，这里通过内部类的方式，实现了自定义的RecordWriter，在MysqlRecordWriter类中实现相关的输出即可完成将reduce结果数据写入到数据库中，具体实现如下图所示：

在MapReduce程序中，在关于job的设置，只需要将输出格式指定为该输出格式即可完成将reduce的结果写入到数据库中。

job.setOutputFormatClass(MysqlOutputFormat.class);

代码实现：

日志一行记录分析类TextLine，该类实现了从日志记录中提取IP信息以及IP次数（一行数据就是1次），代码如下：

/**
* 日志行分析
* @author lulei
*/
package com;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

public class TextLine {
	private String ip;
	private IntWritable one = new IntWritable(1);
	//标识数据是否为可用
	private boolean right = true;
	
	public TextLine(String textLine){
		//检验一行日志数据是否符合要求，如不符合，将其标识为不可用
		if (textLine == null || "".equals(textLine)) {
			this.right = false;
			return;
		}
		String []strs = textLine.split(" ");
		if (strs.length < 2) {
			this.right = false;
			return;
		}
		this.ip = strs[0];
	}
	
	public boolean isRight() {
		return right;
	}

	/**
	 * 返回map的输出key值
	 * @return
	 */
	public Text getIPCountMapOutKey() {
		return new Text(this.ip);
	}
	
	/**
	 * 返回map的输出value值
	 * @return
	 */
	public IntWritable getIPCountMapOutValue() {
		return this.one;
	}
}

IPCountMR类实现了MapReduce功能，实现了日志数据中的IP出现次数统计，代码如下：

/**
 * 各IP出现次数统计
 */
package com;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class IPCountMR extends Configured implements Tool{
	/**
	 * ip个数统计map
	 * @author lulei
	 */
	public static class IPCountMap extends Mapper<LongWritable, Text, Text, IntWritable> {

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			TextLine textLine = new TextLine(value.toString());
			if (textLine.isRight()) {
				context.write(textLine.getIPCountMapOutKey(), textLine.getIPCountMapOutValue());
			}
		}
		
	}
	
	/**
	 * ip个数统计reduce
	 * @author lulei
	 */
	public static class IPCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

		@Override
		public void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable value : values) {
				sum += value.get();
			}
			context.write(key, new IntWritable(sum));
		}
		
	}
	
	@SuppressWarnings("deprecation")
	@Override
	public int run(String[] arg0) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf);
		job.setJobName("ipcount");
		job.setInputFormatClass(TextInputFormat.class);
		
		//将输出设置为MysqlOutputFormat
		job.setOutputFormatClass(MysqlOutputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		job.setMapperClass(IPCountMap.class);
		job.setCombinerClass(IPCountReduce.class);
		job.setReducerClass(IPCountReduce.class);
		
		FileInputFormat.addInputPath(job, new Path(arg0[0]));
		//个人认为下面应该可以不设置的，但是不设置就会报错，不知道是什么地方出了问题
		MysqlOutputFormat.setOutputPath(job,  new Path(arg0[1]));
		
		job.waitForCompletion(true);
		
		return job.isSuccessful() ? 0 : 1;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		try {
			int res = ToolRunner.run(new Configuration(), new IPCountMR(), args);
			System.exit(res);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

MysqlOutputFormat类实现了reduce自定义输出，将reduce计算结果输出到数据库中，代码如下：

package com;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

@SuppressWarnings("hiding")
public class MysqlOutputFormat<Text, IntWritable> extends FileOutputFormat<Text, IntWritable> {
	//Mysql RecordWriter私有类
	private static class MysqlRecordWriter<Text, IntWritable> extends RecordWriter<Text, IntWritable> {
		private LogDB logdb;
		
		/**
		 * 使用外部传进来的LogDB对象
		 * @param logdb
		 */
		MysqlRecordWriter(LogDB logdb){
			this.logdb = logdb;
		}
		@Override
		public void close(TaskAttemptContext arg0) throws IOException,
				InterruptedException {
			// TODO Auto-generated method stub
			
		}

		
		/**
		 * 将key-value写入到数据库中
		 */
		@Override
		public void write(Text key, IntWritable value) throws IOException,
				InterruptedException {
			// TODO Auto-generated method stub
			logdb.insert(key.toString(), value.toString());
		}
		
	}

	@Override
	public RecordWriter<Text, IntWritable> getRecordWriter(
			TaskAttemptContext arg0) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		//返回MysqlRecordWriter对象
		return new MysqlRecordWriter<Text, IntWritable>(new LogDB());
	}

}

上面的LogDB类是自己封装的数据库操作类，实现了数据的插入，具体实现代码如下：

package com;

import java.sql.SQLException;

import com.lulei.db.manager.DBServer;

public class LogDB {
	//新建连接池
	private DBServer dBServer = new DBServer("proxool.log");
	
	/**
	 * 将数据插入至数据库
	 * @param ip
	 * @param num
	 */
	public void insert(String ip, String num) {
		try {
			dBServer.insert("insert into logmp(ip, num) values ('"
					+ ip +"','" +
					num +"')");
		} catch (SQLException e) {
		} finally {
			dBServer.close();
		}
	}
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		new LogDB().insert("127.0.0.2", "1");
	}

}

上面程序中的DBServer类是基于连接池proxool-0.9.1.jar封装的数据库操作类，这里就不做详细的介绍，这里也可以不通过连接池，直接将数据写入到数据库中，这不是本事例的重点，就不做详细的介绍。

上传运行：
具体的操作命令参照博客 http://blog.csdn.net/xiaojimanman/article/details/40184581 中的上传运行。

执行结果：

程序执行结束后，通过命令查看相应的数据表记录，计算结果已经正确写入到数据库中，如下图所示：