Java实现HDFS文本解析写入到Hbase中

package com.xxx.report.service;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.xxx.report.config.Constants;
import com.xxx.report.util.HbaseUtilQA;
import com.xxx.report.util.MD5Util;
import com.xxx.report.util.TimeUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
* @author yangxin-ryanx
*/
public class LockAppLog2Hbase implements Serializable{
private static final Logger LOG = LoggerFactory.getLogger(LockAppLog2Hbase.class);
private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMdd");
private static final String CF_NAME = "cf";
private static final String tableName = "rd:app_log";

public void run(String master, String startTime, String endTime) {
long start = System.currentTimeMillis();
LOG.info("Start run the log parser...");
startTime = startTime.replace("-", "");
endTime = endTime.replace("-","");
SparkConf conf = new SparkConf().setAppName(Constants.SPARK_APP_NAME_YangXin).setMaster(master);
JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
List<String> list = Lists.newArrayList();
Calendar calendar = Calendar.getInstance();
calendar.set(Calendar.YEAR, Integer.valueOf(startTime.substring(0,4)));
calendar.set(Calendar.MONTH, Integer.valueOf(startTime.substring(4,6)) - 1);
calendar.set(Calendar.DATE, Integer.valueOf(startTime.substring(6,8)));
String date = startTime;
while (!date.equals(endTime)){
list.add(date);
calendar.add(Calendar.DATE, 1);
date = simpleDateFormat.format(calendar.getTime());
}
list.add(endTime);

for (String day : list){
StringBuffer path = new StringBuffer();
path.append(Constants.PREFIX_PATH_YangXin).append(day).append("/*/*");

JavaRDD<String> rdd = javaSparkContext.textFile(path.toString());

if (rdd.isEmpty()){
continue;
}
rdd.foreachPartition(new VoidFunction<Iterator<String>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Iterator<String> logs) throws Exception {
while (logs.hasNext()) {

String log = logs.next();
try {
String snValue = log.split(" ")[10];
if (snValue.equals("-")){
continue;
}
handleLog(log);
} catch (Exception e) {
e.printStackTrace();
LOG.error(e.getMessage());
}
}
}
});
HbaseUtilQA.flush(tableName);
}
long end = System.currentTimeMillis();
System.out.println("耗时："+(end-start)/1000+"秒");
}

private void handleLog(String line){
MD5Util md5Util = new MD5Util();
String snValue = line.split(" ")[10];
String timePart1 = line.split(" ")[0];
String timePart2 = line.split(" ")[1];
String time = timePart1 + " " + timePart2;
String rowKey = md5Util.md5Function(snValue, timePart1);
Long timeStamp = TimeUtil.getTimeStamp(time);
Map<String, String> lockAppMap = Maps.newHashMap();
String key = md5Util.md5Key(line) + "_" + timePart2.replace(".", "").replace(":","");
lockAppMap.put(key, line);
HbaseUtilQA.addRecords(tableName, rowKey, timeStamp, CF_NAME.getBytes(), lockAppMap);
}

public static void main(String[] args){
LockAppLog2Hbase lockAppLog2Hbase = new LockAppLog2Hbase();
String master = args[0];
String startTime = args[1];
String endTime = args[2];
lockAppLog2Hbase.run(master, startTime, endTime);
}
}