hadoop之 mr输出到hbase

时间：2023-02-02 22:30:11

1.注意问题：

1.在开发过程中一定要导入hbase源码中的lib库否则出现如下错误

TableMapReducUtil 找不到什么……

2.编码：

import java.io.IOException;

import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Properties;

import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Mapper.Context;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.util.ToolRunner;

import util.PropUtils;

//import java.util.Objects;

/**

 * Created by zzy on 15/11/23.

 *

 * 1、根据HDFS中的数据编写MR程序，将HDFS中数据按需求清洗到HBase中，

 建议方案：

 行键:车牌号

 列：每天的日期

 列内容：时间->地点:时间->地点....

 然后用聚类算法归类指定车牌号车辆的规律路径。

 (1.先将每天的行车路线清洗出来2.聚类)

 create 'carroutes','f'

  得到聚类所要的数据

 create 'traffic_route','f'

 */

public class CarIdMr {

 static class BatchImportMapper extends Mapper<Object,Text,Text,Text> {

  @Override

  protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {

   String line = value.toString();

   String [] splited = line .split(",");

   if (splited.length>10) {

    String dateStr = splited[3];//2014-01-01 22:50:46

    String place = splited[5];

    String carId = splited[2];

//    String  lu = new String("鲁".getBytes(),"unicode");

//    if (carId.contains("鲁")) {

//     carId = carId.replace("鲁", "lu");

//    }

    String str = dateStr + "--" + place;

    context.write(new Text(carId + ":" + dateStr), new Text(str));

   }

  }

 }

 static  class  CarImportReducer extends  TableReducer<Text,Text,ImmutableBytesWritable>{

  public CarImportReducer(){}

//  @Override

  protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

   byte[] row ;

   byte[] family;

   byte[] qualifier;

   byte[] realValue;

   String carIdDate = key.toString();

   StringTokenizer st = new StringTokenizer(carIdDate,":");

   String carId = st.nextToken();

   String date = st.nextToken();

   String  allplace ="";

   for (Text val : values) {

    if(allplace == ""){

     allplace = val.toString();

    }else{

     allplace = allplace+":"+val.toString();

    }

   }

   row = Bytes.toBytes(carId);

   Put put = new Put(row);

   family = Bytes.toBytes("f");

   qualifier = Bytes.toBytes(date);

   realValue = Bytes.toBytes(allplace);

   put.add(family, qualifier, realValue);

   context.write(new ImmutableBytesWritable(row), put);

  }

   }

 private static final String tableName = "traffic_route";

 public   static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

  if(args.length != 2){

   System.out.println("Usage: <inputpath>");

   ToolRunner.printGenericCommandUsage(System.err);

   return ;

  }

  Properties prop = PropUtils.getProp("hbase.properties");//

  Configuration conf = new Configuration();

  conf.set("hbase.zookeeper.quorum", prop.getProperty("hbase.zookeeper.quorum1"));

  conf.set("hbase.rootdir", prop.getProperty("hbase.rootdir"));

  conf.set(TableOutputFormat.OUTPUT_TABLE, prop.getProperty("hbase.table.traffic_route"));

  Job job = Job.getInstance(conf,CarIdMr.class.getSimpleName());

  TableMapReduceUtil.addDependencyJars(job);

  job.setJarByClass(CarIdMr.class);

  job.setMapperClass(BatchImportMapper.class);

  job.setReducerClass(CarImportReducer.class);

  job.setMapOutputKeyClass(Text.class);

  job.setMapOutputValueClass(Text.class);

  job.setNumReduceTasks(Integer.parseInt(args[1]));

//  job.setInputFormatClass(TextInputFormat.class);

  job.setOutputFormatClass(TableOutputFormat.class);

//  FileInputFormat.setInputPaths(job, "hdfs://192.168.122.211:9000/user/hbase/dk/10.txt");

  FileInputFormat.setInputPaths(job, args[0]);

  job.waitForCompletion(true);

//        FileInputFormat.setInputPaths(job,"");

 }

}

.csharpcode, .csharpcode pre
{
font-size: small;
color: black;
font-family: consolas, "Courier New", courier, monospace;
background-color: #ffffff;
/*white-space: pre;*/
}
.csharpcode pre { margin: 0em; }
.csharpcode .rem { color: #008000; }
.csharpcode .kwrd { color: #0000ff; }
.csharpcode .str { color: #006080; }
.csharpcode .op { color: #0000c0; }
.csharpcode .preproc { color: #cc6633; }
.csharpcode .asp { background-color: #ffff00; }
.csharpcode .html { color: #800000; }
.csharpcode .attr { color: #ff0000; }
.csharpcode .alt
{
background-color: #f4f4f4;
width: 100%;
margin: 0em;
}
.csharpcode .lnum { color: #606060; }

相关文章