// 创建表
   @Test
   public void test01() throws Exception{

       // 1) 创建HBase连接对象:

       //Configuration conf = new Configuration();
       Configuration conf = HBaseConfiguration.create();
       conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
       Connection hbConn = ConnectionFactory.createConnection(conf);

       // 2) 从连接对象中获取相关的管理对象: Admin(对表的操作) 和 Table(对表数据操作)

       Admin admin = hbConn.getAdmin();

       // 3) 执行相关的操作

       boolean flag = admin.tableExists(TableName.valueOf("WATER_BILL")); // 为true 表示存在, 否则为不存在

       if(!flag){
           // 说明 表不存在
           //3.1: 创建 表的构建器对象
           TableDescriptorBuilder descBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("WATER_BILL"));


           //3.2: 在构建器 设置表的列族信息
           ColumnFamilyDescriptorBuilder familyDesc = ColumnFamilyDescriptorBuilder.newBuilder("C1".getBytes());
           descBuilder.setColumnFamily(familyDesc.build());

           //3.3: 基于表构建器  构建表基本信息封装对象
           TableDescriptor desc = descBuilder.build();

           admin.createTable(desc);
       }

       // 4) 处理结果集  -- 此步骤只有查询操作

       // 5) 释放资源
       admin.close();
       hbConn.close();

   }

// 添加数据
    @Test
    public void test02() throws Exception{

        //1) 创建Hbase的连接对象
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");

        Connection hbConn = ConnectionFactory.createConnection(conf);
        //2) 通过连接对象 获取相关的管理对象: admin  和 table

        Table table = hbConn.getTable(TableName.valueOf("WATER_BILL"));
        //3) 执行相关的操作:  添加数据

        Put put = new Put(Bytes.toBytes(4944191));

        put.addColumn("C1".getBytes(),"NAME".getBytes(),"登卫红".getBytes());
        put.addColumn("C1".getBytes(),"ADDRESS".getBytes(),"贵州省铜仁市德江县7单元267室".getBytes());
        put.addColumn("C1".getBytes(),"SEX".getBytes(),"男".getBytes());
        put.addColumn("C1".getBytes(),"PAY_TIME".getBytes(),"2020-05-10".getBytes());
       

        table.put(put);

        //4) 处理结果集:  -- 此步骤不需要

        //5) 释放资源
        table.close();
        hbConn.close();


    }

private Connection hbConn;
private Table table;
private Admin admin;
private String tableName = "WATER_BILL";
@Before
public void before() throws Exception{

    //1) 创建Hbase的连接对象
    Configuration conf = HBaseConfiguration.create();

    conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");

    hbConn = ConnectionFactory.createConnection(conf);
    //2) 通过连接对象 获取相关的管理对象: admin  和 table
    admin = hbConn.getAdmin();
    table = hbConn.getTable(TableName.valueOf(tableName));


}

@Test
public void test03(){
    
    // 3) 执行相关的操作
    
    
    //4) 处理结果集
    
}


@After
public  void after() throws Exception{
    //5. 释放资源
    admin.close();
    table.close();
    hbConn.close();
    
}

// 根据rowkey 查询某一条数据
    @Test
    public void test03() throws Exception{

        // 3) 执行相关的操作 get

        Get get = new Get(Bytes.toBytes(4944191));
        Result result = table.get(get); // 一个Result 表示 就是一行数据

        //4) 处理结果集

        //4.1: 获取一行数据中每一个单元格
        List<Cell> listCells = result.listCells();

        //4.2: 遍历单元格, 从单元格获取数据
        for (Cell cell : listCells) {

            // 从单元格可以获取那些内容: rowkey  列族 列名 列值

            //4.2.1 获取rowkey
            /*byte[] rowArray = cell.getRowArray();

            int rowkey = Bytes.toInt(rowArray, cell.getRowOffset(), cell.getRowLength());*/

            byte[] rowBytes = CellUtil.cloneRow(cell);
            int rowkey = Bytes.toInt(rowBytes);

            // 4.2.2 获取 列族
            byte[] familyBytes = CellUtil.cloneFamily(cell);
            String family = Bytes.toString(familyBytes);
            // 4.2.3 获取 列名(列限定符)
            byte[] qualifierBytes = CellUtil.cloneQualifier(cell);
            String qualifier = Bytes.toString(qualifierBytes);
            // 4.2.4 获取 列值
            byte[] valueBytes = CellUtil.cloneValue(cell);
            String value = Bytes.toString(valueBytes);


            System.out.println("rowkey:"+rowkey +"; 列族为:"+family+"; 列名为:"+qualifier+"; 列值为:"+value);

        }


    }

// 删除数据操作:
    @Test
    public  void  test04() throws Exception{

        // 3) 执行相关的操作 : delete  和 deleteall


        Delete delete = new Delete(Bytes.toBytes(4944191));

        //delete.addColumn("C1".getBytes(),"NAME".getBytes());
        delete.addFamily("C1".getBytes());
        table.delete(delete);


        //4) 处理结果集 -- 此步骤不需要

    }

// 删除表操作:
    @Test
    public  void  test05() throws Exception{
        // 3) 执行相关的操作

        boolean flag = admin.isTableEnabled(TableName.valueOf(tableName)); // 判断是否启用表
        if(flag){

            admin.disableTable(TableName.valueOf(tableName)); //如果启用, 将其先禁用
        }

        admin.deleteTable(TableName.valueOf(tableName));

        //4) 处理结果集 -- 此步骤不需要
    }

//scan操作 
@Test
    public void scan() throws Exception{
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
        Connection hbConn = ConnectionFactory.createConnection(conf);
        Table table = hbConn.getTable(TableName.valueOf("WATER_BILL"));

        //命令行的写法 scan 表名 ,{FILTER=>"QualifierFilter(=,'substring:a')"}

        Scan scan = new Scan();
        FilterList filter = new FilterList();
        SingleColumnValueFilter startValueFilter = new SingleColumnValueFilter("C1".getBytes(),"RECORD_DATE".getBytes(), CompareOperator.GREATER_OR_EQUAL,"2020-06-01".getBytes());
        SingleColumnValueFilter endValueFilter = new SingleColumnValueFilter("C1".getBytes(),"RECORD_DATE".getBytes(), CompareOperator.LESS_OR_EQUAL,"2020-07-01".getBytes()) ;
        filter.addFilter(startValueFilter);
        filter.addFilter(endValueFilter);
        scan.setFilter(filter);
        ResultScanner results = table.getScanner(scan);
        for (Result result : results) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String qulifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                if("NAME".equals(qulifier)){
                    String name = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println(name);
                }
            }

        }
        table.close();
        hbConn.close();

    }

package com.hbase.bulkLoad;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class BulkLoadMapper extends Mapper<LongWritable,Text,ImmutableBytesWritable, Put> {
    private ImmutableBytesWritable k2 = new ImmutableBytesWritable();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1. 获取一行数据
        String line = value.toString();

        //2. 判断是否接收到数据
        if(line != null && !"".equals(line.trim())){

            //3. 对这一行数据执行切割操作

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

            //4. 封装 k2 和 v2的数据
            byte[] rowkey = fields[0].getBytes();
            k2.set(rowkey);

            Put v2 = new Put(rowkey);

            v2.addColumn("C1".getBytes(),"code".getBytes(),fields[1].getBytes());
            v2.addColumn("C1".getBytes(),"rec_account".getBytes(),fields[2].getBytes());
            v2.addColumn("C1".getBytes(),"rec_bank_name".getBytes(),fields[3].getBytes());
            v2.addColumn("C1".getBytes(),"rec_name".getBytes(),fields[4].getBytes());
            v2.addColumn("C1".getBytes(),"pay_account".getBytes(),fields[5].getBytes());
            v2.addColumn("C1".getBytes(),"pay_name".getBytes(),fields[6].getBytes());
            v2.addColumn("C1".getBytes(),"pay_comments".getBytes(),fields[7].getBytes());
            v2.addColumn("C1".getBytes(),"pay_channel".getBytes(),fields[8].getBytes());
            v2.addColumn("C1".getBytes(),"pay_way".getBytes(),fields[9].getBytes());
            v2.addColumn("C1".getBytes(),"status".getBytes(),fields[10].getBytes());
            v2.addColumn("C1".getBytes(),"timestamp".getBytes(),fields[11].getBytes());
            v2.addColumn("C1".getBytes(),"money".getBytes(),fields[12].getBytes());


            //5. 写出去
            context.write(k2,v2);

        }


    }
}

package com.hbase.bulkLoad;

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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class BulkLoadDriver extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {

        //1. 创建 Job对象
        Job job = Job.getInstance(super.getConf(), "BulkLoadDriver");
        // 设置提交yarn 必备参数
        job.setJarByClass(BulkLoadDriver.class);
        //2. 设置 天龙八部
        //2.1: 设置 输入类 和输入路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("hdfs://node1:8020/hbase/bulkload/input"));
        //2.2: 设置 mapper类和 map输出k2和v2的类型
        job.setMapperClass(BulkLoadMapper.class);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Put.class);
        //2.3: 设置 shuffle操作: 分区 排序 规约  分组

        //2.7: 设置 reduce类  和 reduce 输出 k3和v3的类型
        job.setNumReduceTasks(0); // 没有reduce

        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Put.class);

        //2.8: 设置输出类, 及输出路径  Hfile
        job.setOutputFormatClass(HFileOutputFormat2.class);
        // 获取 连接对象 和 table对象
        Connection hbConn = ConnectionFactory.createConnection(super.getConf());
        Table table = hbConn.getTable(TableName.valueOf("TRANSFER_RECORD"));

        HFileOutputFormat2.configureIncrementalLoad(job,table,hbConn.getRegionLocator(TableName.valueOf("TRANSFER_RECORD")));

        HFileOutputFormat2.setOutputPath(job, new Path("hdfs://node1:8020/hbase/bulkload/output"));

        //3. 提交任务
        boolean flag = job.waitForCompletion(true);

        return flag ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        int i = ToolRunner.run(conf, new BulkLoadDriver(), args);

        System.exit(i);
    }
}

/**
  * @Author bigdatalearnshare
  */
object App {

  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")

    val sparkSession = SparkSession
      .builder()
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .master("local[*]")
      .getOrCreate()
    
    val rowKeyField = "id"
    
    val df = sparkSession.read.format("json").load("/people.json")

    val fields = df.columns.filterNot(_ == "id").sorted

    val data = df.rdd.map { row =>
      val rowKey = Bytes.toBytes(row.getAs(rowKeyField).toString)

      val kvs = fields.map { field =>
        new KeyValue(rowKey, Bytes.toBytes("hfile-fy"), Bytes.toBytes(field), Bytes.toBytes(row.getAs(field).toString))
      }

      (new ImmutableBytesWritable(rowKey), kvs)
    }.flatMapValues(x => x).sortByKey()
    
    val hbaseConf = HBaseConfiguration.create(sparkSession.sessionState.newHadoopConf())
    hbaseConf.set("hbase.zookeeper.quorum", "linux-1:2181,linux-2:2181,linux-3:2181")
    hbaseConf.set(TableOutputFormat.OUTPUT_TABLE, "hfile")
    val connection = ConnectionFactory.createConnection(hbaseConf)

    val tableName = TableName.valueOf("hfile")

    //没有HBase表则创建
    creteHTable(tableName, connection)

    val table = connection.getTable(tableName)

    try {
      val regionLocator = connection.getRegionLocator(tableName)

      val job = Job.getInstance(hbaseConf)

      job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
      job.setMapOutputValueClass(classOf[KeyValue])

      HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator)

      val savePath = "hdfs://linux-1:9000/hfile_save"
      delHdfsPath(savePath, sparkSession)

      job.getConfiguration.set("mapred.output.dir", savePath)

      data.saveAsNewAPIHadoopDataset(job.getConfiguration)

      val bulkLoader = new LoadIncrementalHFiles(hbaseConf)
      bulkLoader.doBulkLoad(new Path(savePath), connection.getAdmin, table, regionLocator)

    } finally {
      //WARN LoadIncrementalHFiles: Skipping non-directory hdfs://linux-1:9000/hfile_save/_SUCCESS 不影响,直接把文件移到HBASE对应HDFS地址了
      table.close()
      connection.close()
    }

    sparkSession.stop()
  }

  def creteHTable(tableName: TableName, connection: Connection): Unit = {
    val admin = connection.getAdmin

    if (!admin.tableExists(tableName)) {
      val tableDescriptor = new HTableDescriptor(tableName)
      tableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("hfile-fy")))
      admin.createTable(tableDescriptor)
    }
  }

  def delHdfsPath(path: String, sparkSession: SparkSession) {
    val hdfs = FileSystem.get(sparkSession.sessionState.newHadoopConf())
    val hdfsPath = new Path(path)

    if (hdfs.exists(hdfsPath)) {
      //val filePermission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.READ)
      hdfs.delete(hdfsPath, true)
    }
  }
}

public class HBaseTTLTest {
    //1. 创建一个Hbase的表 (设置 版本确界 以及 TTL)
    //2. 向表添加数据, 添加后  进行多次修改操作(产生历史版本)
    //3. 查询数据:  需要将其历史版本全部都查询到

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

        //1. 根据连接工厂获取hbase的连接对象
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        Connection hbConn = ConnectionFactory.createConnection(conf);

        //2. 获取相关的管理对象
        Admin admin = hbConn.getAdmin();

        //3. 执行相关的操作:

        //3.1: 判断表是否存在
        if( !admin.tableExists(TableName.valueOf("day03_hbaseTTL"))){
            // 只要能进来 说明 表不存在

            TableDescriptorBuilder descBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("day03_hbaseTTL"));

            ColumnFamilyDescriptorBuilder familyDesc = ColumnFamilyDescriptorBuilder.newBuilder("C1".getBytes());

            // 设置 版本 确界  和 TTL
            familyDesc.setMinVersions(3);  // 设置 下界
            familyDesc.setMaxVersions(5);  // 设置 上界

            familyDesc.setTimeToLive(30);  // 设置 ttl  有效时间为30s

            ColumnFamilyDescriptor family = familyDesc.build();
            descBuilder.setColumnFamily(family);

            TableDescriptor desc = descBuilder.build();

            //3.2: 创建表
            admin.createTable(desc);
        }


        //3.3:  添加数据操作
        //3.3.1: 获取 table管理对象
        Table table = hbConn.getTable(TableName.valueOf("day03_hbaseTTL"));

        //3.3.2: 执行添加数据操作

        for(int i = 1 ; i<= 2 ; i++){
            Put put = new Put("rk001".getBytes());

            put.addColumn("C1".getBytes(),"NAME".getBytes(),("zhangsan"+i).getBytes());

            table.put(put);
        }

        //3.4: 查询数据
        Get get = new Get("rk001".getBytes());
        get.readAllVersions(); // 获取所有的历史版本
        Result result = table.get(get);

        //Cell[] rawCells = result.rawCells();
        List<Cell> cells = result.listCells();

        for (Cell cell : rawCells) {

            System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
        }


        //4. 释放资源
        table.close();
        admin.close();
        hbConn.close();

    }
}