一、环境配置
这里选择的环境是hadoop-0.20.2和hbase-0.90.4，Hadoop环境配置参看
这里，HBase环境配置请看
这里。
需要注意的是，本文的需求是在Hadoop上跑MapReduce job来分析日志并将结果持久化到HBase，所以，在编译程序时，Hadoop需要用到HBase和Zookeeper包，因此，需要分别将hbase-0.90.4.jar和zookeeper-3.3.2.jar拷贝到Hadoop的lib目录下，具体操作如下：
#cp /root/hbase-0.90.4/hbase-0.90.4.jar /root/hadoop-0.20.2/lib
#cp /root/hbase-0.90.4/lib/zookeeper-3.3.2.jar /root/hadoop-0.20.2/lib
二、实例编写
日志文件xxxlog.txt的内容如下：
version-------------time-----------------id-------rt----filter--------id----rt-----filter
1.0^A2014-03-03 00:00:01^Ad2000^C4^C3040^Bd2001^C7^C0
1.0^A2014-03-03 00:00:01^Ad3000^C4^C3041^Bd2001^C7^C0
同样，需要将此文件放到hdfs目录下，比如：hadoop fs -put /tmp/input。
为持久化在HBase中创建table和family，比如：./hbase shell，create 'xxxlog', 'dsp_filter'。
为了清晰便于扩展，将Maper、Reducer、Driver分开，具体如下：
1、
Maper
#vi xxxLogMaper.java
import java.io.IOException;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
public class xxxLogMaper
extends Mapper {
public final static String CONTROL_A = "^A";
public final static String CONTROL_B = "^B";
public final static String CONTROL_C = "^C";
public final static int PV_TIME = 1;
public final static int DSP_INFO_LIST = 5;
public final static int DSP_ID = 0;
public final static int DSP_FILTER = 2;
public void map(Object key, Text value, Context context) {
try {
System.out.println("\n------------map come on-----------");
System.out.println("\nline=-----------"+value.toString());
String[] line = value.toString().split(CONTROL_A);
String pvtime = "";
System.out.println("\npvtime=-----------"+line[PV_TIME]);
String year = line[PV_TIME].substring(0, 4);
String month = line[PV_TIME].substring(5, 7);
String day = line[PV_TIME].substring(8, 10);
String hour = line[PV_TIME].substring(11, 13);
String minute = "";
int m_tmp = Integer.parseInt(line[PV_TIME].substring(14, 16));
if (m_tmp >= 0 && m_tmp <= 30) {
minute = "00";
} else {
minute = "30";
}
pvtime = year + month + day + hour + minute;
String[] dspInfoList = line[DSP_INFO_LIST].split(CONTROL_B);
String dspid = "";
String dspfilter = "";
Text k = new Text();
IntWritable v = new IntWritable(1);
for(int i=0; i
System.out.println("\n------------map-----------");
System.out.println("\ndspinfo="+dspInfoList[i]);
String[] dspInfo = dspInfoList[i].split(CONTROL_C);
dspid = dspInfo[DSP_ID];
dspfilter = dspInfo[DSP_FILTER];
//key=ddspid^Afilter^Apvtime, value=1
k.set(dspid+CONTROL_A+dspfilter+CONTROL_A+pvtime);
context.write(k, v);
System.out.println("\nkey="+k.toString());
System.out.println("\nvalue="+v.toString());
}
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2、Reducer
import java.io.IOException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.KeyValue;
public class BidLogReducer
extends TableReducer {
public final static String COL_FAMILY = "dsp_filter";
public final static String COL_NAME = "sum";
private final static String ZK_HOST = "localhost";
private final static String TABLE_NAME = "xxxlog";
public void reduce(Text key, Iterable values, Context context)
throws IOException, InterruptedException {
System.out.println("\n------------reduce come on-----------");
String k = key.toString();
IntWritable v = new IntWritable();
int sum = 0;
for (IntWritable val:values) {
sum += val.get();
}
//v.set(sum);
//context.write(key, v);
System.out.println("\n------------reduce-----------");
System.out.println("\ncur-key="+key.toString());
System.out.println("\ncur-value="+sum);
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum.", ZK_HOST);
HTablePool pool = new HTablePool(conf, 3);
HTable table = (HTable)pool.getTable(TABLE_NAME);
Get getrow = new Get(k.getBytes());
Result r = table.get(getrow);
int m_tmp = 0;
for(KeyValue kv:r.raw()) {
System.out.println("\nraw-KeyValugge---"+kv);
System.out.println("\nraw-row=>"+Bytes.toString(kv.getRow()));
System.out.println("\nraw-family=>"+Bytes.toString(kv.getFamily()));
System.out.println("\nraw-qualifier=>"+Bytes.toString(kv.getQualifier()));
System.out.println("\nraw-value=>"+Bytes.toString(kv.getValue()));
m_tmp += Integer.parseInt(Bytes.toString(kv.getValue()));
}
sum = sum + m_tmp;
v.set(sum);
System.out.println("\nreal-key="+key.toString());
System.out.println("\nreal-value="+v.toString());
Put putrow = new Put(k.getBytes());
putrow.add(COL_FAMILY.getBytes(), COL_NAME.getBytes(), String.valueOf(v).getBytes());
try {
context.write(new ImmutableBytesWritable(key.getBytes()), putrow);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
3、Driver
#vi xxxLogDriver.java
#vi xxxLogReducer.java
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class xxxLogDriver {
public final static String ZK_HOST = "localhost";
public final static String TABLE_NAME = "xxxlog";
public static void main(String[] args) throws Exception {
//Hbase Configuration
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum.", ZK_HOST);
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: please input args");
System.exit(2);
}
Job job = new Job(conf,"xxxLog");
job.setJarByClass(xxxLogDriver.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.out.println("\n------------driver come on-----------");
job.setMapperClass(xxxLogMaper.class);
job.setReducerClass(xxxLogReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
TableMapReduceUtil.initTableReducerJob(TABLE_NAME, xxxLogReducer.class, job);
System.exit(job.waitForCompletion(true)? 0 : 1);
}
}
三、编译运行
在当前目录下编译源码，具体如下：
#javac -classpath /root/hadoop-0.20.2/hadoop-0.20.2-core.jar:/root/hadoop-0.20.2/lib/commons-cli-1.2.jar:/root/hbase-0.90.4/hbase-0.90.4.jar -d ./ xxxLogMaper.java xxxLogReducer.java xxxLogDriver.java
需要注意的是，必须三个一起编译否则出错：
xxxLogDriver.java:22: error: cannot find symbol
job.setMapperClass(xxxLogMaper.class);
打包class文件，具体如下：
#jar cvf xxxLog.jar *class
#rm -rf *class
运行任务，具体如下：
#hadoop jar xxxLog.jar xxxLogDriver /tmp/input /tmp/output
查询结果，具体如下：
#./hbase shell
hbase(main):014:0>scan 'xxxlog'
关于更多Hadoop和HBase的信息请参见
这里，，
这里，
这里，，还有
这里。
阅读(552) | 评论(0) | 转发(0) |
