SIMONE
posts - 495,comments - 227,trackbacks - 0
在HBase里使用MapReduce例子
http://jeffxie.blog.51cto.com/1365360/305538

我在Hadoop的用户邮件列表中看到一些国内的用 户在讯问一些关于如何操作的HBase的问题，还看到了HBase中没有Example。觉得有 必要跟大家分享自己的经验。
在下面的例子中我们分析Apache的log并把这些log进行分析并把分析完 的结果按用户IP为ROW，把log中用户的访问时间，请求方法，用户请求的协议，用户的浏览器，服务状态等写到HBase的表中。


首先我们要在HBase中建立我们的一个表来存储数据。  
  1. public static void creatTable(String table) throws IOException{
  2.             HConnection conn = HConnectionManager.getConnection(conf);
  3.             HBaseAdmin admin = new HBaseAdmin(conf);
  4.             if(!admin.tableExists(new Text(table))){
  5.               System.out.println("1. " + table + " table creating ... please wait");
  6.               HTableDescriptor tableDesc = new HTableDescriptor(table);
  7.               tableDesc.addFamily(new HColumnDescriptor("http:"));
  8.               tableDesc.addFamily(new HColumnDescriptor("url:"));
  9.               tableDesc.addFamily(new HColumnDescriptor("referrer:"));
  10.               admin.createTable(tableDesc);
  11.             } else {
  12.               System.out.println("1. " + table + " table already exists.");
  13.             }
  14.             System.out.println("2. access_log files fetching using map/reduce");
  15.   }
复制代码

然后我们运行一个MapReduce任务来取得log中的每一行 数据。因为我们只要取得数据而不需要对结果进行规约，我们只要编写一个Map程序即可。   
  1. public static class MapClass extends MapReduceBase implements
  2.       Mapper<WritableComparable, Text, Text, Writable> {

  4.     @Override
  5.     public void configure(JobConf job) {
  6.       tableName = job.get(TABLE, "");
  7.     }

  9.     public void map(WritableComparable key, Text value,
  10.         OutputCollector<Text, Writable> output, Reporter reporter)
  11.         throws IOException {
  12.       try {
  13.              AccessLogParser log = new AccessLogParser(value.toString());
  14.         if(table==null)
  15.                 table = new HTable(conf, new Text(tableName));
  16.         long lockId = table.startUpdate(new Text(log.getIp()));
  17.         table.put(lockId, new Text("http:protocol"), log.getProtocol().getBytes());
  18.         table.put(lockId, new Text("http:method"), log.getMethod().getBytes());
  19.         table.put(lockId, new Text("http:code"), log.getCode().getBytes());
  20.         table.put(lockId, new Text("http:bytesize"), log.getByteSize().getBytes());
  21.         table.put(lockId, new Text("http:agent"), log.getAgent().getBytes());
  22.         table.put(lockId, new Text("url:" + log.getUrl()), log.getReferrer().getBytes());
  23.         table.put(lockId, new Text("referrer:" + log.getReferrer()), log.getUrl().getBytes());

  25.         table.commit(lockId, log.getTimestamp());
  26.       } catch (ParseException e) {
  27.         e.printStackTrace();
  28.       } catch (Exception e) {
  29.         e.printStackTrace();
  30.       }
  31.     }
  32.   }
复制代码

我们在Map程序中对于传进来的每一行先交给AccessLogParser去处理在AccessLogParser德构造器中用一个正则表达式"([^ ]*) ([^ ]*) ([^ ]*) \\[([^]]*)\\] \"([^\"]*)\"　" ([^ ]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\".*"来匹配每一行的log。接下来我们把这些AccessLogParser处理出来的结果更新到HBase的表中去，好的， 我们的程序写完了。我们要启动一个MapReduce的话我们要对工作进行配置。   
  1. public static void runMapReduce(String table,String dir) throws IOException{
  2.           Path tempDir = new Path("log/temp");
  3.           Path InputDir = new Path(dir);
  4.           FileSystem fs = FileSystem.get(conf);
  5.           JobConf jobConf = new JobConf(conf, LogFetcher.class);
  6.           jobConf.setJobName("apache log fetcher");
  7.           jobConf.set(TABLE, table);
  8.           Path[] in = fs.listPaths(InputDir);
  9.           if (fs.isFile(InputDir)) {
  10.               jobConf.setInputPath(InputDir);
  11.           } else {
  12.               for (int i = 0; i < in.length; i++) {
  13.                 if (fs.isFile(in[i])) {
  14.                   jobConf.addInputPath(in[i]);
  15.                 } else {
  16.                   Path[] sub = fs.listPaths(in[i]);
  17.                   for (int j = 0; j < sub.length; j++) {
  18.                     if (fs.isFile(sub[j])) {
  19.                       jobConf.addInputPath(sub[j]);
  20.                     }
  21.                   }
  22.                 }
  23.               }
  24.             }
  25.             jobConf.setOutputPath(tempDir);
  26.             jobConf.setMapperClass(MapClass.class);

  28.             JobClient client = new JobClient(jobConf);
  29.             ClusterStatus cluster = client.getClusterStatus();
  30.             jobConf.setNumMapTasks(cluster.getMapTasks());
  31.             jobConf.setNumReduceTasks(0);

  33.             JobClient.runJob(jobConf);
  34.             fs.delete(tempDir);
  35.             fs.close();
  36.   }
复制代码

在上面的代码中我们先产生一个jobConf对象，然后设定我们的InputPath和OutputPath，告诉MapReduce我们的Map类，设 定我们用多少个Map任务和Reduce任务，然后我们不任务提交给JobClient，关于MapReduce跟详细的资料在Hadoop Wiki上。
下载：源码和已编译好的jar文件example-src.tgz
例子的运行命令是：

bin/hadoop jar examples.jar logfetcher <access_log file or directory> <table_name>

如何运行上面的应用程序呢？我们假定解压缩完Hadoop分发包的目录为%HADOOP%
拷贝%HADOOP%\contrib\hbase\bin下的文件到%HADOOP%\bin下,拷贝%HADOOP%\contrib\hbase \conf的文件到%HADOOP%\conf下,拷贝%HADOOP%\src\contrib\hbase\lib的文件到%HADOOP%\lib 下,拷贝%HADOOP%\src\contrib\hbase\hadoop-*-hbase.jar的文件到%HADOOP%\lib下.然后编辑配 置文件hbase-site.xml设定你的hbase.master例子：192.168.2.92:60000。把这些文件分发到运行Hadoop的 机器上去。在regionservers文件添加上这些已分发过的地址。运行bin/start-hbase.sh命令启动HBase，把你的 apache log文件拷贝到HDFS的apache-log目录下，等启动完成后运行下面的命令。

bin/hadoop jar examples.jar logfetcher apache-log apache

访问http://localhost:50030/能 看到你的MapReduce任务的运行情况，访问http://localhost:60010/能 看到HBase的运行情况。

hbaseguiinterface.jpg

等任务MapReduce完成后访问http://localhost:60010/hql.jsp,在Query输入框中输入 SELECT * FROM apache limit=50;。将会看到已经插入表中的数据。 hqlguiinterface.jpg
posted on 2013-02-22 14:12 SIMONE 阅读(2492) 评论(0)  编辑  收藏 所属分类: hbase

只有注册用户登录后才能发表评论。
网站导航:
 
Copyright ©2025 SIMONE Powered By博客园 模板提供：沪江博客