HBase使用

@Test
public void createNamespace() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Admin admin = connection.getAdmin();
  NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("bigdata").build();
  admin.createNamespace(namespaceDescriptor);
  admin.close();
  connection.close();
}

@Test
public void tableExists() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Admin admin = connection.getAdmin();
  boolean result = admin.tableExists(TableName.valueOf("bigdata:person"));
  System.out.println("table exists " + result);
  admin.close();
  connection.close();
}

@Test
public void createTable() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Admin admin = connection.getAdmin();
  TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("bigdata", "student"));
  ColumnFamilyDescriptorBuilder familyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"));
  familyDescriptorBuilder.setMaxVersions(3);
  tableDescriptorBuilder.setColumnFamily(familyDescriptorBuilder.build());
  admin.createTable(tableDescriptorBuilder.build());
  admin.close();
  connection.close();
}

@Test
public void dropTable() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Admin admin = connection.getAdmin();
  admin.disableTable(TableName.valueOf("bigdata", "student"));
  admin.deleteTable(TableName.valueOf("bigdata", "student"));
  admin.close();
  connection.close();
}

@Test
public void addData() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Table stuTable = connection.getTable(TableName.valueOf("bigdata", "student"));
  Put put = new Put(Bytes.toBytes("1001"));
  put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan"));
  stuTable.put(put);
  stuTable.close();
  connection.close();
}

@Test
public void delData() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Table stuTable = connection.getTable(TableName.valueOf("bigdata", "student"));
  Delete delete = new Delete(Bytes.toBytes("1001"));
  stuTable.delete(delete);
  stuTable.close();
  connection.close();
}

@Test
public void scanData() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Table stuTable = connection.getTable(TableName.valueOf("bigdata", "student"));
  Scan scan = new Scan();
  ResultScanner scanner = stuTable.getScanner(scan);
  for (Result result : scanner) {
    Cell[] cells = result.rawCells();
    for (Cell cell : cells) {
      System.out.println("行键:" + Bytes.toString(CellUtil.cloneRow(cell)));
      System.out.println("列族:" + Bytes.toString(CellUtil.cloneFamily(cell)));
      System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
      System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell)));
    }
  }
  stuTable.close();
  connection.close();
}

@Test
public void getRow() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Table stuTable = connection.getTable(TableName.valueOf("bigdata", "student"));
  Get get = new Get(Bytes.toBytes("1001"));
  //设置读取所有版本
  //get.readAllVersions();
  //设置时间戳
  //get.setTimestamp()
  Result result = stuTable.get(get);
  for (Cell cell : result.rawCells()) {
    System.out.println("行键：" + Bytes.toString(result.getRow()));
    System.out.println("列族：" + Bytes.toString(CellUtil.cloneFamily(cell)));
    System.out.println("列：" + Bytes.toString(CellUtil.cloneQualifier(cell)));
    System.out.println("值：" + Bytes.toString(CellUtil.cloneValue(cell)));
    System.out.println("时间戳：" + cell.getTimestamp());
  }
  stuTable.close();
  connection.close();
}

@Test
public void getRowQualifier() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Table stuTable = connection.getTable(TableName.valueOf("bigdata", "student"));
  Get get = new Get(Bytes.toBytes("1001"));
  get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));
  Result result = stuTable.get(get);
  for (Cell cell : result.rawCells()) {
    System.out.println("行键：" + Bytes.toString(result.getRow()));
    System.out.println("列族：" + Bytes.toString(CellUtil.cloneFamily(cell)));
    System.out.println("列：" + Bytes.toString(CellUtil.cloneQualifier(cell)));
    System.out.println("值：" + Bytes.toString(CellUtil.cloneValue(cell)));
    System.out.println("时间戳：" + cell.getTimestamp());
  }
  stuTable.close();
  connection.close();
}

public class FruitMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    context.write(key, value);
  }
}


public class FruitReducer extends TableReducer<LongWritable, Text, NullWritable> {

  private String inputColumnFamilyColumnNamesArg;
  private List<ColumnFamilyColumnName> columnFamilyColumnNames = new ArrayList<>();

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    Configuration configuration = context.getConfiguration();
    this.inputColumnFamilyColumnNamesArg = configuration.get("columnFamily-columnName");
    String[] inputColumnFamilyColumnNamesArgArray = inputColumnFamilyColumnNamesArg.split(",");
    for (String item : inputColumnFamilyColumnNamesArgArray) {
      String[] columnFamilyColumnName = item.split(":");
      columnFamilyColumnNames.add(new ColumnFamilyColumnName(columnFamilyColumnName[0], columnFamilyColumnName[1]));
    }
    for (ColumnFamilyColumnName columnFamilyColumnName : columnFamilyColumnNames) {
      System.out.println("family: " + columnFamilyColumnName.getFamily());
      System.out.println("column: " + columnFamilyColumnName.getColumnName());
    }
  }

  @Override
  protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    for (Text value : values) {
      String[] fields = value.toString().split("\t");
      System.out.println("length :" + fields.length);
      Put put = new Put(Bytes.toBytes(fields[0]));
      int fieldIndex = 1;
      for (ColumnFamilyColumnName columnFamilyColumnName : columnFamilyColumnNames) {
        if (fieldIndex < fields.length && fieldIndex <= columnFamilyColumnNames.size()) {
          System.out.println(fields[fieldIndex]);
          put.addColumn(
            Bytes.toBytes(columnFamilyColumnName.getFamily()),
            Bytes.toBytes(columnFamilyColumnName.getColumnName()),
            Bytes.toBytes(fields[fieldIndex]));
        }
        fieldIndex++;
      }
      context.write(NullWritable.get(), put);
    }
  }

  @Data
  @AllArgsConstructor
  private static class ColumnFamilyColumnName {
    private String family;
    private String columnName;
  }
}

public class FruitDriver implements Tool {

  private Configuration configuration;

  @Override
  public int run(String[] args) throws Exception {
    // 1.获取Job对象
    Job job = Job.getInstance(configuration);
    // 2.获取驱动类路径
    job.setJarByClass(FruitDriver.class);
    // 3.设置Mapper&Mapper输出类型
    job.setMapperClass(FruitMapper.class);
    job.setMapOutputKeyClass(LongWritable.class);
    job.setMapOutputValueClass(Text.class);
    // 4.设置Reducer
    TableMapReduceUtil.initTableReducerJob("bigdata:student", FruitReducer.class, job);
    // 5.设置输入路径
    FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.22.161:9000/data"));
    // 6.提交任务
    job.submit();
    return job.waitForCompletion(true) ? 0 : -1;
  }

  @Override
  public void setConf(Configuration conf) {
    configuration = conf;
  }

  @Override
  public Configuration getConf() {
    return this.configuration;
  }

  public static void main(String[] args) throws Exception {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.22.161:9000");
    configuration.set("columnFamily-columnName", "info:name,info:sex");
    configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
    int res = ToolRunner.run(configuration, new FruitDriver(), args);
    System.exit(res);
  }
}

/**
  * 预分区
  */
@Test
public void preSplit() throws IOException {
  Configuration configuration = new Configuration();
  configuration.set(HConstants.ZOOKEEPER_QUORUM, "centos161,centos162,centos163");
  Connection connection = ConnectionFactory.createConnection(configuration);
  Admin admin = connection.getAdmin();
  byte[][] splitKeys = new byte[][]{Bytes.toBytes("30000"), Bytes.toBytes("60000")};
  TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(
    TableName.valueOf(NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR, "staff"));
  ColumnFamilyDescriptorBuilder familyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"));
  tableDescriptorBuilder.setColumnFamily(familyDescriptorBuilder.build());
  admin.createTable(tableDescriptorBuilder.build(), splitKeys);
  admin.close();
  connection.close();
}