正文

Hbase 表的Rowkey设计避免数据热点

IP云
IP云V管理员/4 阅读
12/01
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

一、案例分析

常见避免数据热点问题的处理方式有：加盐、哈希、反转等方法结合预分区使用。

由于目前原数据第一字段为时间戳形式，第二字段为电话号码，直接存储容易引起热点问题，通过加随机列、组合时间戳、字段反转的方式来设计Rowkey，来实现既能高效查询又能避免热点问题。

二、代码部分 

 package beifeng.hadoop.hbase;

 import java.io.IOException;

 import java.text.SimpleDateFormat;

 import java.util.Date;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.conf.Configured;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.hbase.HBaseConfiguration;

 import org.apache.hadoop.hbase.HColumnDescriptor;

 import org.apache.hadoop.hbase.HTableDescriptor;

 import org.apache.hadoop.hbase.MasterNotRunningException;

 import org.apache.hadoop.hbase.TableName;

 import org.apache.hadoop.hbase.ZooKeeperConnectionException;

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

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

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

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

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

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.NullWritable;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.Mapper;

 import org.apache.hadoop.mapreduce.Reducer;

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

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 import org.apache.hadoop.util.Tool;

 import org.apache.hadoop.util.ToolRunner;

 import org.apache.hadoop.io.Text;

 /**

  * 遵循rowkey的设计原则

  *  1.rowkey不能过长

  *  2.唯一性，加随机列  md5

  *  3.注意避免产生数据热点

  *  4.满足更多的查询场景

  * @author Administrator

  *

  */

 public class LoadData extends Configured implements Tool {

     /**

      * 综合考虑 使用时间和手机 做组合key，能更好的满足应用场景

      * @author Administrator

      *

      */

     public static class LoadDataMapper extends Mapper<LongWritable, Text, LongWritable, Text> {

         //专门处理时间戳 =》标准时间格式

         SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHsss");

         private Text mapOutputValue = new Text();

         @Override

         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, LongWritable, Text>.Context context)

                 throws IOException, InterruptedException {

             String line = value.toString();

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

             //将切分的第一个字段转成标准时间

             String formatDate = sdf.format(new Date(Long.parseLong(splited[0].trim())));

             //将手机号码反转

             String phoneNumber = splited[1].toString();

                         String reversePhoneNumber = new StringBuffer(phoneNumber).reverse().toString();

             String rowKeyString = reversePhoneNumber +"|"+formatDate;

             //反转手机号+“|”+时间 +正行内容拼接

             mapOutputValue.set(rowKeyString+"\t"+ line);

             context.write(key, mapOutputValue);

         }

     }

     public static class LoadDataReuducer extends TableReducer<LongWritable, Text, NullWritable>{

         //设置HBase的列簇

             private static final String COLUMN_FAMAILY = "info";

         @Override

         protected void reduce(LongWritable key, Iterable<Text> values,

                 Reducer<LongWritable, Text, NullWritable, Mutation>.Context context)

                 throws IOException, InterruptedException {

             for (Text value:values) {

                 String[] splited = value.toString().split("\t");

                 String rowKey = splited[0];

             //    System.err.println(rowKey);

                 Put put = new Put(rowKey.getBytes());

                 //put.addColumn(COLUMN_FAMAILY.getBytes(),"row".getBytes(),value.getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "reportTime".getBytes(), splited[1].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "apmac".getBytes(), splited[3].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "acmac".getBytes(), splited[4].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "host".getBytes(), splited[5].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "siteType".getBytes(), splited[6].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "upPackNum".getBytes(), splited[7].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "downPackNum".getBytes(), splited[8].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "unPayLoad".getBytes(), splited[9].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(), "downPayLoad".getBytes(), splited[10].getBytes());

                 put.add(COLUMN_FAMAILY.getBytes(),"httpStatus".getBytes(),splited[11].getBytes());

                 context.write(NullWritable.get(), put);

             }

         }

     }

     public static void createTable(String tableName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException {

         Configuration  conf = HBaseConfiguration.create();

         conf.set("hbase.zookeeper.quorum", "beifeng01");

         HBaseAdmin admin = new HBaseAdmin(conf);

         TableName tName = TableName.valueOf(tableName);

         HTableDescriptor htd = new HTableDescriptor(tName);

         HColumnDescriptor hcd = new HColumnDescriptor("info");

         htd.addFamily(hcd);

         if(admin.tableExists(tName)) {

             System.out.println(tableName+"is exist,trying to recrate the table");

             admin.disableTable(tName);

             admin.deleteTable(tName);

         }

         admin.createTable(htd);

         System.out.println("create new table"+ " " + tableName);

     }

     public int run(String[] args) throws Exception {

     Configuration conf = this.getConf();

         conf.set("hbase.zookeeper.quorum", "beifeng01");

         conf.set(TableOutputFormat.OUTPUT_TABLE, "phoneLog"); 

         createTable("phoneLog"); 

         Job job = Job.getInstance(conf, this.getClass().getSimpleName());

         job.setJarByClass(this.getClass());

         job.setNumReduceTasks(1); 

         // map class

         job.setMapperClass(LoadDataMapper.class);

         job.setMapOutputKeyClass(LongWritable.class);

         job.setMapOutputValueClass(Text.class);  

         // reduce class

         job.setReducerClass(LoadDataReuducer.class);

         job.setOutputFormatClass(TableOutputFormat.class); 

         Path inPath = new Path(args[0]);

         FileInputFormat.addInputPath(job, inPath);

        boolean isSucced = job.waitForCompletion(true);

        return isSucced ? 0 : 1;

     }

     public static void main(String[] args) throws Exception {

         Configuration conf = HBaseConfiguration.create(); 

         //指定HDFS数据地址

         args = new String[] {"hdfs://hbase/data/input/HTTP_20130313143750.data"};

            int status = ToolRunner.run(

                    conf,

                    new LoadData(),

                    args);

     System.exit(status);

     }

 }

运行完程序后scan 查看效果 

hbase(main):004:0> scan 'phoneLog', {LIMIT => 2}

ROW                                COLUMN+CELL

 01787706731|2013031314048         column=info:acmac, timestamp=1544022103345, value=120.196.100.82

 01787706731|2013031314048         column=info:apmac, timestamp=1544022103345, value=00-FD-07-A4-7B-08:CMCC

 01787706731|2013031314048         column=info:downPackNum, timestamp=1544022103345, value=2

 01787706731|2013031314048         column=info:downPayLoad, timestamp=1544022103345, value=120

 01787706731|2013031314048         column=info:host, timestamp=1544022103345, value=

 01787706731|2013031314048         column=info:httpStatus, timestamp=1544022103345, value=200

 01787706731|2013031314048         column=info:reportTime, timestamp=1544022103345, value=1363157988072

 01787706731|2013031314048         column=info:siteType, timestamp=1544022103345, value=

 01787706731|2013031314048         column=info:unPayLoad, timestamp=1544022103345, value=120

 01787706731|2013031314048         column=info:upPackNum, timestamp=1544022103345, value=2

 10007032831|2013031314045         column=info:acmac, timestamp=1544022103345, value=120.196.100.99

 10007032831|2013031314045         column=info:apmac, timestamp=1544022103345, value=20-7C-8F-70-68-1F:CMCC

 10007032831|2013031314045         column=info:downPackNum, timestamp=1544022103345, value=3

 10007032831|2013031314045         column=info:downPayLoad, timestamp=1544022103345, value=180

 10007032831|2013031314045         column=info:host, timestamp=1544022103345, value=

 10007032831|2013031314045         column=info:httpStatus, timestamp=1544022103345, value=200

 10007032831|2013031314045         column=info:reportTime, timestamp=1544022103345, value=1363157985079

 10007032831|2013031314045         column=info:siteType, timestamp=1544022103345, value=

 10007032831|2013031314045         column=info:unPayLoad, timestamp=1544022103345, value=360

 10007032831|2013031314045         column=info:upPackNum, timestamp=1544022103345, value=6

相关阅读

目录[+]