Hadoop简单的Map/Reduce

爱她吗

/**
* 需求:
*     统计Hbase数据库中某个字符串的使用人数
*     字符串存储格式 1002;1003,2003,1443;1232,232
*   最后统计 模块   使用人数
*       1    1002    X
*       ...
*     把最后的统计信息存储到结果表中.
*/
public class CountUse {
public static Logger loger = Wloger.loger;
/**
* Internal Mapper to be run by Hadoop.
*/
public static class MapTask extends TableMapper<UdcSellHTable, IntWritable> {
private UdcSellHTable udcSell = new UdcSellHTable();
private String tableName;
private byte[] family;
private HashMap<String, byte[]> indexes;
private UdcSellHTable word = new UdcSellHTable();
private final static IntWritable one = new IntWritable(1);
/**
* Map 统计使用人数
*/
@SuppressWarnings("static-access")
@Override
protected void map(ImmutableBytesWritable rowKey, Result result,
Context context) throws IOException, InterruptedException {
// 搜索数据库设置到对象当中
for (Map.Entry<String, byte[]> entity : indexes.entrySet()) {
String column = entity.getKey();
PropertyName property = new PropertyName(family, entity
.getValue());
Object type = udcSell.types.get(column);
Object value = CountUseUtil.changeValue(new Entity(tableName,
result), property, type);// 数据库中的值
try {
loger.info("debug:"+udcSell);
loger.info("debug:"+UdcSellHTable.filter(column));
loger.info("debug:"+value);
BeanUtils.setProperty(udcSell, UdcSellHTable.filter(column),
value);
} catch (Exception e) {
loger.info("对象属性定义有问题，请check[" + udcSell + ":" + column
+ "[" + value + "]]");
}
}
// filter 属性过滤，过滤掉不符合规则的记录
String widget = udcSell.getWidget();// widget数据,//674324;321,321,312;3321
word.setWidget(widget);//设置widget
loger.debug("debug:"+widget);
List<String> keys = CountUseUtil.getKeys(widget);
for (String key : keys) {
word.setUserId(udcSell.getUserId());
word.setDate(udcSell.getDate());
word.setModuleId(Long.parseLong(key));
word.setUseCount(1);
loger.info("Map <word>" + word.toString() + "</word>");
context.write(word, one);
}
}
@SuppressWarnings("static-access")
@Override
protected void setup(Context context) throws IOException,
InterruptedException {
Configuration configuration = context.getConfiguration();
tableName = configuration.get("input.tablename");
String familyName = configuration.get("input.familyname");
udcSell.setDate(CountUseUtil.yestoday());
String[] fields = new String[] { udcSell.COLUMN_USERID,
udcSell.getWidgetDate() };
indexes = new HashMap<String, byte[]>();
for (String field : fields) {
indexes.put(field, Bytes.toBytes(field));
}
family = Bytes.toBytes(familyName);
}
}
/**
* Reduce进行合计
*
*/
public static class CombinerTask extends
Reducer<UdcSellHTable, IntWritable, UdcSellHTable, IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(UdcSellHTable key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
loger.info("combiner <key>" + key + "</key>");
loger.info("combiner <sum>" + sum + "</sum>");
context.write(key, result);
}
}
/**
* Reduce合计入库
*
*/
public static class Reduce extends
TableReducer<UdcSellHTable, IntWritable, ImmutableBytesWritable> {
private UdcSellHTable udcSell = new UdcSellHTable();
private String tableName;
private IntWritable result = new IntWritable();
private HashMap<String, byte[]> indexes;
private byte[] family;
public void reduce(UdcSellHTable key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
key.setUseCount(result.get());
loger.info("Reduce <key>" + key + "</key>");
loger.info("Reduce <sum>" + sum + "</sum>");
String keyStr = "1_" + key.getModuleId();
Put put = new Put(Bytes.toBytes(keyStr));// 以模块id为key
for (Map.Entry<String, byte[]> entity : indexes.entrySet()) {
try {
put.add(family, entity.getValue(), Bytes.toBytes(BeanUtils
.getProperty(key, entity.getKey())));
} catch (Exception e) {
loger.info("对象属性定义有问题，请check[" + udcSell + ":"
+ entity.getKey() + "[" + entity.getValue() + "]]");
}
}
ImmutableBytesWritable imw = new ImmutableBytesWritable(Bytes
.toBytes(tableName));
context.write(imw, put);
}
@SuppressWarnings("static-access")
@Override
protected void setup(Context context) throws IOException,
InterruptedException {
Configuration configuration = context.getConfiguration();
tableName = configuration.get("output.tablename");
String familyName = configuration.get("output.familyname");
family = Bytes.toBytes(familyName);
String[] fields = new String[] { udcSell.COLUMN_USERID,udcSell.COLUMN_MODULEID,
udcSell.COLUMN_USE };
indexes = new HashMap<String, byte[]>();
for (String field : fields) {
indexes.put(field, Bytes.toBytes(field));
}
}
}
/**
*
*
* Job configuration.
*/
public static Job configureJob(Configuration conf, String[] args)
throws IOException {
String inputTable = "udc_sell";
String inputFamily = "s_year";
String outputTable = "job_result";
String outputFamily = "s_base";
conf.set("input.tablename", inputTable);
conf.set("input.familyname", inputFamily);
conf.set(TableInputFormat.INPUT_TABLE, inputTable);
conf.set("output.tablename", outputTable);
conf.set("output.familyname", outputFamily);
conf.set(TableOutputFormat.OUTPUT_TABLE, outputTable);
Job job = new Job(conf, inputTable);
job.setNumReduceTasks(1);
job.setJarByClass(CountUse.class);
job.setMapperClass(MapTask.class);
job.setMapOutputKeyClass(UdcSellHTable.class);
job.setMapOutputValueClass(IntWritable.class);
job.setCombinerClass(CombinerTask.class);
job.setReducerClass(Reduce.class);
job.setInputFormatClass(TableInputFormat.class);
job.setOutputFormatClass(MultiTableOutputFormat.class);
return job;
}
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
loger.info("开始 job");
Job job = configureJob(conf, otherArgs);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}