spark读HFile对hbase表数据进行分析

要求：计算hasgj表，计算每天新增mac数量。

因为spark直接扫描hbase表，对hbase集群访问量太大，给集群造成压力，这里考虑用spark读取HFile进行数据分析。

1、建立hasgj表的快照表：hasgjSnapshot

语句为：snapshot 'hasgj','hasgjSnapshot'

2、计算每天mac增量的代码如下：

package com.ba.sparkReadHbase.operatorHfile.hfileinputformat;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.NavigableMap;

import java.util.Set;

import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HBaseConfiguration;

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

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

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

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

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

import org.apache.hadoop.hbase.protobuf.ProtobufUtil;

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;

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

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

import org.apache.hadoop.mapreduce.Job;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.Function;

import scala.Tuple2;

public class SparkReadHFile {

    private static String convertScanToString(Scan scan) throws IOException {

        ClientProtos.Scan proto = ProtobufUtil.toScan(scan);

        return Base64.encodeBytes(proto.toByteArray());

    }

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

        final String date=args[0];

        int max_versions = 3;

        SparkConf sparkConf = new SparkConf().setAppName("sparkReadHfile");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        Configuration hconf = HBaseConfiguration.create();

        hconf.set("hbase.rootdir", "/hbase");

        hconf.set("hbase.zookeeper.quorum", "master,slave1,slave2");

        Scan scan = new Scan();

        scan.addFamily(Bytes.toBytes("ba"));

        scan.setMaxVersions(max_versions);

        hconf.set(TableInputFormat.SCAN, convertScanToString(scan));

        Job job = Job.getInstance(hconf);

        Path path = new Path("/snapshot");

        String snapName ="hasgjSnapshot";

        TableSnapshotInputFormat.setInput(job, snapName, path);

        JavaPairRDD<ImmutableBytesWritable, Result> newAPIHadoopRDD = sc.newAPIHadoopRDD(job.getConfiguration(), TableSnapshotInputFormat.class, ImmutableBytesWritable.class,Result.class);

        List<String> collect = newAPIHadoopRDD.map(new Function<Tuple2<ImmutableBytesWritable, Result>, String>(){

            private static final long serialVersionUID = 1L;

            public String call(Tuple2<ImmutableBytesWritable, Result> v1)

                    throws Exception {

                // TODO Auto-generated method stub

                String newMac =null;

                Result result = v1._2();

                if (result.isEmpty()) {

                    return null;

                }

                String rowKey = Bytes.toString(result.getRow());

                //System.out.println("行健为："+rowKey);

                NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes("ba"));

                Set<Entry<byte[], byte[]>> entrySet = familyMap.entrySet();

                java.util.Iterator<Entry<byte[], byte[]>> it = entrySet.iterator();

                String colunNmae =null;

                String minDate="34561213";

                while(it.hasNext()){

                    colunNmae = new String(it.next().getKey());//列

                    if(colunNmae.compareTo(minDate)<0){

                        minDate=colunNmae;

                    }

                }

                if (date.equals(minDate)) {

//                    row=rowKey.substring(4);

                    newMac=rowKey;

                    //ls.add(rowKey.substring(4));

                    //bf.append(rowKey+"----");

                }

                return  newMac;

            }

         }).collect();

          ArrayList<String> arrayList = new ArrayList<String>();

            for (int i = 0; i < collect.size(); i++) {

                if (collect.get(i) !=null) {

                    arrayList.add(collect.get(i));

                }

            }

             System.out.println("新增mac数"+(arrayList.size()));

    }

}

3、特别说明：

hasgj表的表结构：

0000F470ABF3A587                          column=ba:20170802, timestamp=1517558687930, value=
0000F470ABF3A587                          column=ba:20170804, timestamp=1517593923254, value=
0000F470ABF3A587                          column=ba:20170806, timestamp=1517620990589, value=
0000F470ABF3A587                          column=ba:20170809, timestamp=1517706294758, value=
0000F470ABF3A587                          column=ba:20170810, timestamp=1517722369020, value=
0000F470ABF3A587                          column=ba:20170811, timestamp=1517796060984, value=
0000F470ABF3A587                          column=ba:20170816, timestamp=1517882948856, value=
0000F470ABF3A587                          column=ba:20170818, timestamp=1517912603602, value=
0000F470ABF3A587                          column=ba:20170819, timestamp=1517938488763, value=
0000F470ABF3A587                          column=ba:20170821, timestamp=1517989742180, value=
0000F470ABF3A587                          column=ba:20170827, timestamp=1518383470292, value=
0000F470ABF3A587                          column=ba:20170828, timestamp=1520305841272, value=
0000F470ABF3A587                          column=ba:20170831, timestamp=1522115116459, value=
0000F4730088A5D3                          column=ba:20170805, timestamp=1517598564121, value=
0000F47679E83F7D                          column=ba:20170817, timestamp=1517890046587, value=
0000F47FBA753FC7                          column=ba:20170827, timestamp=1518365792130, value=
0000F48C02F8EB83                          column=ba:20170810, timestamp=1517729864592, value=
0000F49578E63F55                          column=ba:20170828, timestamp=1520302223714, value=
0000F4AC4A93F7A5                          column=ba:20170810, timestamp=1517724545955, value=
0000F4B4807679AA                          column=ba:20170801, timestamp=1517543775374, value=
0000F4B7E374C0FF                          column=ba:20170804, timestamp=1517578239073, value=
0000F4BDBF6EBF37                          column=ba:20170829, timestamp=1520558747936, value=
0000F4CB52FDDA58                          column=ba:20170806, timestamp=1517638015583, value=
0000F4CB52FDDA58                          column=ba:20170807, timestamp=1517677405900, value=

4、提交作业命令：

./spark-submit --master yarn-client --num-executors 7 --executor-cores 2 --driver-memory 2g --executor-memory 30g --class com.ba.sparkReadHbase.operatorHfile.hfileinputformat.SparkReadHFile /home/xxx0108/ftttttttt/testJar/sparkTest9.jar 20170806

巴特西

spark读HFile对hbase表数据进行分析

最新文章

热门文章