hive处理日志，自定义inputformat

开放环境，hadoop-0.20.2，hive-0.6

1.日志分隔符
Xml代码 复制代码 收藏代码
  1. 2010-05-31 10:50:17|||61.132.4.82|||http://www.360buy.com/product/201185.html  

分隔符是“ ||| ”，这是为了尽可能防止日志正文出现与分隔符相同的字符而导致数据混淆。
hive 的内部分隔符是“ 01 ”，所以我们需要做一下转换

2.编写自定义InputFormat
Java代码 复制代码 收藏代码
  1. package com.jd.cloud.clickstore;  
  2.   
  3. import java.io.IOException;  
  4.   
  5. import org.apache.hadoop.io.LongWritable;  
  6. import org.apache.hadoop.io.Text;  
  7. import org.apache.hadoop.mapred.FileSplit;  
  8. import org.apache.hadoop.mapred.InputSplit;  
  9. import org.apache.hadoop.mapred.JobConf;  
  10. import org.apache.hadoop.mapred.JobConfigurable;  
  11. import org.apache.hadoop.mapred.RecordReader;  
  12. import org.apache.hadoop.mapred.Reporter;  
  13. import org.apache.hadoop.mapred.TextInputFormat;  
  14.   
  15. /** 
  16.  * 自定义hadoop的 org.apache.hadoop.mapred.InputFormat 
  17.  *  
  18.  * @author winston 
  19.  *  
  20.  */  
  21. public class ClickstreamInputFormat extends TextInputFormat implements  
  22.         JobConfigurable {  
  23.   
  24.     public RecordReader<LongWritable, Text> getRecordReader(  
  25.             InputSplit genericSplit, JobConf job, Reporter reporter)  
  26.             throws IOException {  
  27.   
  28.         reporter.setStatus(genericSplit.toString());  
  29.         return new ClickstreamRecordReader(job, (FileSplit) genericSplit);  
  30.     }  
  31. }  

3.自定义ClickstreamRecordReader实现RecordReader接口，并重写next方法
 
Java代码 复制代码 收藏代码
  1. /** Read a line. */  
  2.   public synchronized boolean next(LongWritable key, Text value)  
  3.     throws IOException {  
  4.   
  5.     while (pos < end) {  
  6.       key.set(pos);  
  7.   
  8.       int newSize = in.readLine(value, maxLineLength,  
  9.                                 Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),  
  10.                                          maxLineLength));  
  11.         
  12.       //start  
  13.       String strReplace = value.toString().toLowerCase().replaceAll("\|\|\|" , "01" );  
  14.       Text txtReplace = new Text();  
  15.       txtReplace.set(strReplace );  
  16.       value.set(txtReplace.getBytes(), 0, txtReplace.getLength());  
  17.       //end  
  18.         
  19.         
  20.       if (newSize == 0) {  
  21.         return false;  
  22.       }  
  23.       pos += newSize;  
  24.       if (newSize < maxLineLength) {  
  25.         return true;  
  26.       }  
  27.   
  28.       // line too long. try again  
  29.       LOG.info("Skipped line of size " + newSize + " at pos " + (pos - newSize));  
  30.     }  
  31.   
  32.     return false;  
  33.   }  

我们可以直接使用LineRecordReader，修改next方法

3.启动hive，添加我们自己刚刚添加的类


4.创建数据库
Java代码 复制代码 收藏代码
  1. create table clickstream_table(time string, ip string, url string) stored as INPUTFORMAT 'com.jd.cloud.clickstore.ClickstreamInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION '/data/clickstream_20110216.txt';  


5.导入数据
Java代码 复制代码 收藏代码
  1. LOAD DATA LOCAL INPATH '/data/clickstream_20110216.txt' OVERWRITE INTO TABLE clickstream_table;  


6.查询刚刚到入的数据
select * from clickstream_table;



参考http://wiki.apache.org/hadoop/Hive/SerDe
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/java20130722/p/3206914.html