= 目的 = 
 This program will parse your apache log and store it into Hbase.
 

 = 如何使用 = 
 * 1. Upload apache logs ( /var/log/apache2/access.log* ) to hdfs (default: /user/waue/apache-log) \

{{{ 
$ bin/hadoop dfs -put /var/log/apache2/ apache-log
}}}
 * 2. parameter "dir" in main contains the logs.

 * 3. you should filter the exception contents manually, 
{{{ ex:  ::1 - - [29/Jun/2008:07:35:15 +0800] "GET / HTTP/1.0" 200 729 "...
  }}}
}}}
= 結果 = 
 1 執行以下指令
{{{
	hql > select * from apache-log;

}}}
 2 結果

{{{
+-------------------------+-------------------------+-------------------------+

| Row                     | Column                  | Cell                    |

+-------------------------+-------------------------+-------------------------+

| 118.170.101.250         | http:agent              | Mozilla/4.0 (compatible;|

|                         |                         |  MSIE 4.01; Windows 95) |

+-------------------------+-------------------------+-------------------------+

| 118.170.101.250         | http:bytesize           | 318                     |

+-------------------------+-------------------------+-------------------------+

..........(skip)........

+-------------------------+-------------------------+-------------------------+

| 87.65.93.58             | http:method             | OPTIONS                 |

+-------------------------+-------------------------+-------------------------+

| 87.65.93.58             | http:protocol           | HTTP/1.1                |

+-------------------------+-------------------------+-------------------------+

| 87.65.93.58             | referrer:-              | *                       |

+-------------------------+-------------------------+-------------------------+

| 87.65.93.58             | url:*                   | -                       |

+-------------------------+-------------------------+-------------------------+

31 row(s) in set. (0.58 sec)


}}}

 = LogParser.java = 

{{{
package tw.org.nchc.code;


import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.Locale;

import java.util.StringTokenizer;

import java.util.regex.Matcher;

import java.util.regex.Pattern;




public class LogParser {

  private String ip;

  private String protocol;

  private String method;

  private String url;

  private String code;

  private String byteSize;

  private String referrer;

  private String agent;

  private long timestamp;


  private static Pattern p = Pattern

  .compile("([^ ]*) ([^ ]*) ([^ ]*) \\[([^]]*)\\] \"([^\"]*)\"" +

                  " ([^ ]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\".*");

  public LogParser(String line) throws ParseException, Exception{

	 Matcher matcher = p.matcher(line);

	 if(matcher.matches()){

		 this.ip = matcher.group(1);

		 // IP address of the client requesting the web page.

		 if(isIpAddress(ip)){

			 SimpleDateFormat sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z",Locale.US);

			 this.timestamp = sdf.parse(matcher.group(4)).getTime();

			 String[] http = matcher.group(5).split(" ");

			 this.method = http[0];

			 this.url = http[1];

			 this.protocol = http[2];

			 this.code = matcher.group(6);

			 this.byteSize = matcher.group(7);

			 this.referrer = matcher.group(8);

			 this.agent = matcher.group(9);

		 }

	 }



  }


  public static boolean isIpAddress(String inputString) {

    StringTokenizer tokenizer = new StringTokenizer(inputString, ".");

    if (tokenizer.countTokens() != 4) {

      return false;

    }

    try {

      for (int i = 0; i < 4; i++) {

        String t = tokenizer.nextToken();

        int chunk = Integer.parseInt(t);

        if ((chunk & 255) != chunk) {

          return false;

        }

      }

    } catch (NumberFormatException e) {

      return false;

    }

    if (inputString.indexOf("..") >= 0) {

      return false;

    }

    return true;

  }


  public String getIp() {

    return ip;

  }


  public String getProtocol() {

    return protocol;

  }


  public String getMethod() {

    return method;

  }


  public String getUrl() {

    return url;

  }


  public String getCode() {

    return code;

  }


  public String getByteSize() {

    return byteSize;

  }


  public String getReferrer() {

    return referrer;

  }


  public String getAgent() {

    return agent;

  }


  public long getTimestamp() {

    return timestamp;

  }


}
}}}

 = LogParserGo.java = 

{{{
/**
 * Program: LogFetcher.java
 * Editor: Waue Chen 
 * From :  NCHC. Taiwn
 * Last Update Date: 07/02/2008
 */

package tw.org.nchc.code;


import java.io.IOException;


import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HBaseAdmin;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTable;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.Writable;

import org.apache.hadoop.io.WritableComparable;

import org.apache.hadoop.mapred.ClusterStatus;

import org.apache.hadoop.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reporter;



// import AccessLogParser

/**

 * Access_log fetcher. TODO: FgnStatLog, Error_log, Access_log (Default,

 * W3CExtended, IISw3cExtended)

 */

public class LogParserGo {

	static HBaseConfiguration conf = new HBaseConfiguration();



	public static final String TABLE = "table.name";



	static String tableName;



	static HTable table = null;



//	static boolean eclipseRun = false;

	static void print(String str){

		System.out.println("STR  = "+str);

	}

	public static class MapClass extends MapReduceBase implements

			Mapper<WritableComparable, Text, Text, Writable> {



		@Override

		public void configure(JobConf job) {

			tableName = job.get(TABLE, "");

		}



		public void map(WritableComparable key, Text value,

				OutputCollector<Text, Writable> output, Reporter reporter)

				throws IOException {

			

			try {

				/*

				print(value.toString());

				FileWriter out = new FileWriter(new File(

				"/home/waue/mr-result.txt"));

				out.write(value.toString());

				out.flush();

				out.close();

				*/

				LogParser log = new LogParser(value.toString());

				

				if (table == null)

					table = new HTable(conf, new Text(tableName));

				long lockId = table.startUpdate(new Text(log.getIp()));

				table.put(lockId, new Text("http:protocol"), log.getProtocol()

						.getBytes());

				table.put(lockId, new Text("http:method"), log.getMethod()

						.getBytes());

				table.put(lockId, new Text("http:code"), log.getCode()

						.getBytes());

				table.put(lockId, new Text("http:bytesize"), log.getByteSize()

						.getBytes());

				table.put(lockId, new Text("http:agent"), log.getAgent()

						.getBytes());

				table.put(lockId, new Text("url:" + log.getUrl()), log

						.getReferrer().getBytes());

				table.put(lockId, new Text("referrer:" + log.getReferrer()),

						log.getUrl().getBytes());

				table.commit(lockId, log.getTimestamp());

				

			} catch (Exception e) {

				e.printStackTrace();

			}

			

		}

	}



	// do it to resolve warning : FileSystem.listPaths

	static public Path[] listPaths(FileSystem fsm, Path path)

			throws IOException {

		FileStatus[] fss = fsm.listStatus(path);

		int length = fss.length;

		Path[] pi = new Path[length];

		for (int i = 0; i < length; i++) {

			pi[i] = fss[i].getPath();

		}

		return pi;

	}



	public static void runMapReduce(String table, String dir)

			throws IOException {

		Path tempDir = new Path("/tmp/Mylog/");

		Path InputDir = new Path(dir);

		FileSystem fs = FileSystem.get(conf);

		JobConf jobConf = new JobConf(conf, LogParserGo.class);

		jobConf.setJobName("apache log fetcher");

		jobConf.set(TABLE, table);

		Path[] in = listPaths(fs, InputDir);

		if (fs.isFile(InputDir)) {

			jobConf.setInputPath(InputDir);

		} else {

			for (int i = 0; i < in.length; i++) {

				if (fs.isFile(in[i])) {

					jobConf.addInputPath(in[i]);

				} else {

					Path[] sub = listPaths(fs, in[i]);

					for (int j = 0; j < sub.length; j++) {

						if (fs.isFile(sub[j])) {

							jobConf.addInputPath(sub[j]);

						}

					}

				}

			}

		}

		jobConf.setOutputPath(tempDir);



		jobConf.setMapperClass(MapClass.class);



		JobClient client = new JobClient(jobConf);

		ClusterStatus cluster = client.getClusterStatus();

		jobConf.setNumMapTasks(cluster.getMapTasks());

		jobConf.setNumReduceTasks(0);



		JobClient.runJob(jobConf);



		fs.delete(tempDir);

		fs.close();

	}



	public static void creatTable(String table) throws IOException {

		HBaseAdmin admin = new HBaseAdmin(conf);

		if (!admin.tableExists(new Text(table))) {

			System.out.println("1. " + table

					+ " table creating ... please wait");

			HTableDescriptor tableDesc = new HTableDescriptor(table);

			tableDesc.addFamily(new HColumnDescriptor("http:"));

			tableDesc.addFamily(new HColumnDescriptor("url:"));

			tableDesc.addFamily(new HColumnDescriptor("referrer:"));

			admin.createTable(tableDesc);

		} else {

			System.out.println("1. " + table + " table already exists.");

		}

		System.out.println("2. access_log files fetching using map/reduce");
	}
	public static void main(String[] args) throws IOException {
		String table_name = "apache-log2";
		String dir = "/user/waue/apache-log";
		creatTable(table_name);
		runMapReduce(table_name, dir);
	}
}

}}}