HBase的操作命令

古城微笑少年丶 2021-12-19 01:25 330阅读 0赞

前文：

一、控制台命令

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70][]

二、Java的API

package cn.tedu.hbase;
    
    import java.util.ArrayList;
    import java.util.Iterator;
    import java.util.List;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.HColumnDescriptor;
    import org.apache.hadoop.hbase.HTableDescriptor;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.Get;
    import org.apache.hadoop.hbase.client.HBaseAdmin;
    import org.apache.hadoop.hbase.client.HTable;
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.client.Result;
    import org.apache.hadoop.hbase.client.ResultScanner;
    import org.apache.hadoop.hbase.client.Scan;
    import org.junit.Test;
    
    public class HbaseDemo {
    
    	@Test
    	public void connectAndCreate() throws Exception {
    
    		Configuration conf = HBaseConfiguration.create();
    		conf.set("hbase.zookeeper.quorum", "hadoop01:2181,hadoop02:2181");
    
    		// HBase管理者:创建删除表
    		HBaseAdmin admin = new HBaseAdmin(conf);
    
    		// create 'tb2',{NAME=>'cf1',VERSIONS=>3},'cf2'
    		HTableDescriptor table = new HTableDescriptor(TableName.valueOf("tb2"));
    		HColumnDescriptor cf1 = new HColumnDescriptor("cf1".getBytes());
    		cf1.setMaxVersions(3);
    		HColumnDescriptor cf2 = new HColumnDescriptor("cf2".getBytes());
    		table.addFamily(cf1);
    		table.addFamily(cf2);
    
    		// 执行
    		admin.createTable(table);
    		admin.close();
    	}
    
    	@Test
    	public void putData() throws Exception {
    
    		Configuration conf = HBaseConfiguration.create();
    		conf.set("hbase.zookeeper.quorum", "hadoop01:2181");
    
    		HTable table = new HTable(conf, "tb2");
    		List<Put> puts = new ArrayList<>();
    		for (int i = 0; i < 100; i++) {
    			// put 'tb2','row1','cf1:num','1'
    			Put put = new Put(("row" + i).getBytes());
    			put.add("cf1".getBytes(), "num".getBytes(), (i + "").getBytes());
    			puts.add(put);
    		}
    
    		// 执行
    		table.put(puts);
    		table.close();
    
    	}
    
    	@Test
    	public void getData() throws Exception {
    		Configuration conf = HBaseConfiguration.create();
    		conf.set("hbase.zookeeper.quorum", "hadoop01:2181");
    
    		// get 'tb2','row1',"cf1:num"
    		HTable table = new HTable(conf, "tb2");
    		Get get = new Get("row1".getBytes());
    
    		// 执行
    		Result result = table.get(get);
    		byte[] value = result.getValue("cf1".getBytes(), "num".getBytes());
    
    		System.out.println(new String(value));
    		table.close();
    	}
    
    	@Test
    	public void scanData() throws Exception {
    		Configuration conf = HBaseConfiguration.create();
    		conf.set("hbase.zookeeper.quorum", "hadoop01:2181");
    
    		// scan 'tb2',{COLUMNS=>['cf1:num']}
    		HTable table = new HTable(conf, "tb2");
    		Scan scan = new Scan();
    		// scan.addFamily("cf1".getBytes())
    		scan.addColumn("cf1".getBytes(), "num".getBytes());
    		scan.setStartRow("row1".getBytes());
    		scan.setStopRow("row5".getBytes());
    
    		// 执行
    		ResultScanner rs = table.getScanner(scan);
    		Iterator<Result> it = rs.iterator();
    
    		while (it.hasNext()) {
    			Result result = it.next();
    			byte[] num = result.getValue("cf1".getBytes(), "num".getBytes());
    			System.out.println(new String(num));
    		}
    		table.close();
    	}
    
    }

三、Scala的API

package cn.tedu
    
    import org.apache.hadoop.hbase.HBaseConfiguration
    import org.apache.hadoop.hbase.client.Put
    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.mapred.TableOutputFormat
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.protobuf.ProtobufUtil
    import org.apache.hadoop.hbase.util.Base64
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.hadoop.mapred.JobConf
    import org.apache.spark.SparkConf
    import org.apache.spark.SparkContext
    import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
    import org.apache.hadoop.hbase.filter.RowFilter
    import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
    import org.apache.hadoop.hbase.filter.RegexStringComparator
    
    object hbase_test {
    
      def main(args: Array[String]): Unit = {
        //1.创建sc
        val conf = new SparkConf();
        conf.setAppName("spark_hbase_test_scan")
        conf.setMaster("local[5]")
        val sc = new SparkContext(conf);
    
        //2.创建配置信息
        val hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", "hadoop01,hadoop02");
        hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");
        hbaseConf.set(TableInputFormat.INPUT_TABLE, "tb2")
        //3.创建设置scan
        val scan = new Scan();
        scan.setStartRow("row1".getBytes)
        scan.setStopRow("row5".getBytes)
        //scan.setFilter(new RowFilter(CompareOp.EQUAL,new RegexStringComparator("3")));
        hbaseConf.set(TableInputFormat.SCAN, Base64.encodeBytes(ProtobufUtil.toScan(scan).toByteArray()));
        //4.读取数据
        val rdd = sc.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
        //5.遍历结果获取信息
        rdd.foreach(t => {
          val r = t._2
          val rk = Bytes.toString(r.getRow())
          println(rk)
        })
        //insertHBase();
        //queryHBase();
      }
    
      def queryHBase() {
        //1.创建sc
        val conf = new SparkConf();
        conf.setAppName("spark_hbase_test_query")
        conf.setMaster("local[5]")
        val sc = new SparkContext(conf)
    
        //2.创建配置信息
        val hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", "hadoop01,hadoop02");
        hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");
        hbaseConf.set(TableInputFormat.INPUT_TABLE, "tb2");
    
        //3.读取数据
        val rdd = sc.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
    
        //4.遍历结果获取信息
        rdd.foreach(t => {
          val r = t._2
          val rk = Bytes.toString(r.getRow)
          val num = Bytes.toString(r.getValue("cf1".getBytes, "num".getBytes))
          println(rk + "~" + num)
        })
      }
    
      def insertHBase() {
        //0.配置序列化器
        System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
    
        //1.创建sc
        val conf = new SparkConf();
        conf.setAppName("spark_hbase_test_insert")
        conf.setMaster("local[5]")
        val sc = new SparkContext(conf);
    
        //2.创建配置信息
        val hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", "hadoop01,hadoop02");
        hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");
        val jobConf = new JobConf(hbaseConf);
        jobConf.setOutputFormat(classOf[TableOutputFormat]);
        jobConf.set(TableOutputFormat.OUTPUT_TABLE, "tb2");
    
        //3.准备数据
        val put1 = new Put("row_1".getBytes);
        put1.add("cf1".getBytes, "num".getBytes, "101".getBytes)
    
        val put2 = new Put("row_2".getBytes);
        put2.add("cf1".getBytes, "num".getBytes, "102".getBytes)
    
        val rdd = sc.makeRDD(Array((new ImmutableBytesWritable, put1), (new ImmutableBytesWritable, put2)))
    
        //4.将rdd中的数据写入hbase
        rdd.saveAsHadoopDataset(jobConf);
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70]: /images/20211219/54a8f3bb99c349e4a01ce0ae6c902385.png