分布式数据库HBase开发： Java API 管理表

1. javaAPI获取表的列表
2. 修改表
3. 禁用表，启用表，删除表

前言：

        第一关用一个foreach遍历，输出三行代码就行。

        第二关对照左边教程，对号入座就行。

        第三关删除（记得先停用再删除，不然报错），停用，启用都还比较简单，创建表函数添加列的AIP有一个是过时的（HColumnDescriptor）。

第一关要：

第一关答案：

package step1;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.*;

public class Task {

	public void showTableList() throws Exception {
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象
		List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
		for(TableDescriptor tableDescriptor: tableDescriptors){
			System.out.println("Table：" + tableDescriptor.getTableName());
			System.out.println("\texists：" + admin.tableExists(tableDescriptor.getTableName()));
			System.out.println("\tenabled：" + admin.isTableEnabled(tableDescriptor.getTableName()));
		}
		



		/********* End *********/
	}
}

 第二关要求：

第二关答案：

	package step2;

import java.io.IOException;


import org.apache.hadoop.conf.*;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.*;

public class Task {
	
	public void updateTables()throws Exception{
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象
		TableName tableName1 = TableName.valueOf("t_emp2");
		TableName tableName2 = TableName.valueOf("t_dept2");
				ColumnFamilyDescriptorBuilder buildFamily = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("data"));//创建builder对象
				buildFamily.setBlocksize(1024*1024);//设置缓存大小
				buildFamily.setBlockCacheEnabled(false);//设置关闭缓存，默认是true
				buildFamily.setCompressionType(Compression.Algorithm.GZ);//默认是NONE
				ColumnFamilyDescriptor family = buildFamily.build();//构建Family对象
				admin.modifyColumnFamily(tableName1, family);//调用修改方法，方法接收两个参数：TableName，ColumnFamilyDescriptor
				admin.deleteColumnFamily(tableName1, Bytes.toBytes("data1"));//删除表中名为data的列族    
			
			
				ColumnFamilyDescriptorBuilder buildFamily1 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("data1"));//创建builder对象
				buildFamily1.setMinVersions(2);
				buildFamily1.setMaxVersions(5);
				buildFamily1.setInMemory(true); //默认是false
				buildFamily1.setTimeToLive(60*60*24); //以秒为单位，超过这个时间设置的就会在下一次大合并中被删除
				ColumnFamilyDescriptor family1 = buildFamily1.build();//构建Family对象
				admin.modifyColumnFamily(tableName2, family1);//调用修改方法，方法接收两个参数：TableName，ColumnFamilyDescriptor

				admin.deleteColumnFamily(tableName2, Bytes.toBytes("data"));//删除表中名为data的列族  
		
			







		/********* End *********/
	}
	
}

 第三关要求：

第三关答案：

package step3;

import java.io.IOException;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.*;


public class Task {

	/**
	 * 刪除表
	 * @param tableName 表名
	 * @throws Exception
	 */
	public void deleteTable(String tableName)throws Exception{
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象
		TableName testName = TableName.valueOf(Bytes.toBytes(tableName));
		admin.disableTable(testName);
		admin.deleteTable(testName);
	


		/********* End *********/
	}
	
	/**
	 * 创建表
	 * @param tableName  表名
	 * @param columnNames 列族的动态数组
	 * @throws Exception
	 */
	public void createTable(String tableName,String... columnNames)throws Exception{
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象

		// TableName tableName1 = TableName.valueOf(Bytes.toBytes(tableName));//定义表名
		// HTableDescriptor htd = new HTableDescriptor(tableName1);//定义表对象
		// for(String s: columnNames){
		// 	htd.addFamily(new HColumnDescriptor(Bytes.toBytes(s)));
		// }

        //构建一个Test_teacher_info表
        TableDescriptorBuilder test_teacher_info = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));

        for(String s: columnNames){
			ColumnFamilyDescriptor of = ColumnFamilyDescriptorBuilder.of(s);
 			test_teacher_info.setColumnFamily(of);
		}
    
        //构建
        TableDescriptor build = test_teacher_info.build();
 
        admin.createTable(build);



		/********* End *********/
	}
	
	/**
	 * 启用表
	 * @param tableName 
	 * @throws Exception 
	 */
	public void enableTable(String tableName) throws Exception{
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象
		TableName demoName = TableName.valueOf(Bytes.toBytes(tableName));
		admin.enableTable(demoName);


		/********* End *********/
	}
	
	/**
	 * 禁用表
	 * @param tableName
	 */
	public void disableTable(String tableName)throws Exception{
		/********* Begin *********/
		Configuration conf = HBaseConfiguration.create(); //使用create()静态方法就可以得到Configuration对象
		Connection conn = ConnectionFactory.createConnection(conf); //config为前文的配置对象
		Admin admin = conn.getAdmin(); //使用连接对象获取Admin对象
		TableName testName = TableName.valueOf(Bytes.toBytes(tableName));
		admin.disableTable(testName);


		/********* End *********/
	}
	
}

1、Java API概述

API（Application Programing Interface）：应用程序编程接口，是java提供的一些预定义的函数（或者黑匣子）。目的：基于API实现程序的快速编写，在无需了解源码基础之上，即可完成功能开发。

JavaAPI通常是一些基础、通用的基础类库，通过这些基础的类库，可以构造复杂的功能。

JavaAPI在线文档；java1.6 java1.7 java1.8

1.1、如何学习

针对一个API首先看概述，了解类的作用，然后看构造函数，了解类的创建方式。最后看方法，了解提供的功能和实现。

1.2、如何调用

使用import语句，可调用Java类库（Java API）的方法。也或者直接写API的方法。使用编辑器导入包的快捷方式，引用API的对应方法。

import导入包有两种语法形式：

• 单类型导入(single-type-import)， 例如 import java.io.File;
• 按需类型导入(type-import-on-demand)，例如 import java.io.*;

第二种按需类型导入，虽然结尾是*，给人感觉是导入io包下所有文件，其实不是。它也仅仅是导入当前类下所引用的方法。尽量使用单类型导入。可以提高编译速度，并且避免包类型冲突。

2、常见的API

• java.lang  提供利用java编程语言进行程序设计的基础类。（核心类库）不用import可以直接调用。
• java  until  包含集合框架，遗留的collection类，事件模型，日期和时间设施，国际化和各种实用工具类。
• java io 通过数据流，序列化和文件系统提供系统输入和输出。
• java applet 提供创建applet 所必须的的类和applet 用来和其他applet 上下文通信的类。
• java awt 包含用于创建用户界面和绘制图形图像的所有的类。
• java beans  包含与开发beans 有关的类，即 基于javaBeans 架构的组件。
• java math 提供用于执行任意精度整数算法（BigInteger）和任意精度小数算法（BigDecimal）的类。
• java net 为实现网络应用程序提供的类。
• java sql  提供使用java 编程语言访问并处理存储在数据源（通常指关系型数据库）中的数据API。

关于Elasticsearch Java API Client客户端如何连接以及如何对索引和文档进行基本的增删改查操作请查看我的上一篇博文：Elasticsearch RestHighLevelClient 已标记为被弃用 它的替代方案 Elasticsearch Java API Client 的基础教程及迁移方案_无枫的博客，分享Java及Vue方向的技术文章-CSDN博客

 本篇主要描述在Elasticsearch Java API Client客户端中如何进行批量操作，以及如何进行各种条件及类型的查询（Search）方法。

1. 批量添加文档

// 创建客户端连接部分
RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200)).build();
ElasticsearchTransport transport = new RestClientTransport(
        restClient, new JacksonJsonpMapper());
ElasticsearchClient client = new ElasticsearchClient(transport);

// 构建一个批量操作BulkOperation的集合
List<BulkOperation> bulkOperations = new ArrayList<>();
// 向集合添加数据
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("zhangsan", "男", 30)).id("3001").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("lisi", "女", 30)).id("3002").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu", "男", 40)).id("3003").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu1", "女", 40)).id("3004").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu2", "男", 50)).id("3005").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu3", "男", 50)).id("3006").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu33", "男", 50)).id("3007").index("newapi")).build());
bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(new Test("wangwu333", "男", 50)).id("3008").index("newapi")).build());
// 使用bulk方法执行批量操作并获得响应
BulkResponse response = client.bulk(e->e.index("newapi").operations(bulkOperations));
// 打印结果
System.out.println(response.took());
System.out.println(response.items());

// 关闭客户端连接部分
transport.close();
restClient.close();

首先需要创建ES客户端连接，然后构建一个批量操作BulkOperation的ArrayList集合，并向其添加你需要插入的文档数据，这里新版客户端可以直接传入Java对象，ES会在内部自行处理。

使用ES客户端的bulk方法进行批量操作并获得批量操作的响应结果，最后打印出结果即可。

2. 批量删除文档

// 创建ES客户端部分
RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200)).build();
ElasticsearchTransport transport = new RestClientTransport(
        restClient, new JacksonJsonpMapper());
ElasticsearchClient client = new ElasticsearchClient(transport);

// 构建批量操作对象BulkOperation的集合
List<BulkOperation> bulkOperations = new ArrayList<>();

// 向集合中添加需要删除的文档id信息
bulkOperations.add(new BulkOperation.Builder().delete(d-> d.index("newapi").id("3001")).build());
bulkOperations.add(new BulkOperation.Builder().delete(d-> d.index("newapi").id("3002")).build());
bulkOperations.add(new BulkOperation.Builder().delete(d-> d.index("newapi").id("3003")).build());

// 调用客户端的bulk方法，并获取批量操作响应结果
BulkResponse response = client.bulk(e->e.index("newapi").operations(bulkOperations));
System.out.println(response.took());
System.out.println(response.items());

// 关闭ES客户端部分
transport.close();
restClient.close();

与批量添加文档类似，首先需要创建ES客户端，同样使用BulkOperation集合来存储批量操作的内容，不同的是这次使用BulkOperationBuilder的delete方法构建批量删除操作，最后调用ES客户端的bulk方法执行，获取的响应结果同样为BulkResponse类型。

3. 分页查询

// 分页查询
SearchResponse<Test> response3 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .matchAll(m -> m)
                )
                .from(4)
                .size(2)
        , Test.class);
System.out.println(response3.took());
System.out.println(response3.hits().total().value());
response3.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的分页查询主要使用SearchResponse的from和size方法传入参数，其中from代表数据开始的下表位置，size代表每次查询需要获取到的文档数量。

4. 查询排序

// 查询排序
SearchResponse<Test> response4 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .matchAll(m -> m)
                )
                .sort(sort -> sort
                        .field(f -> f
                                .field("age")
                                .order(SortOrder.Desc)
                        )
                )
        , Test.class);
System.out.println(response4.took());
System.out.println(response4.hits().total().value());
response4.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的查询排序主要使用sort方法传入排序参数，我这里使用了lambda形式传入参数。与RestAPI一致，需要传入field名称以及排序方式，如ASC、DESC。

5. 过滤字段

// 过滤字段
SearchResponse<Test> response5 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .matchAll(m -> m)
                )
                .sort(sort -> sort
                        .field(f -> f
                                .field("age")
                                .order(SortOrder.Desc)
                        )
                )
                .source(source -> source
                                .filter(f -> f
                                  .includes("name")
                                  .excludes("")
                                )
                )
        , Test.class);
System.out.println(response5.took());
System.out.println(response5.hits().total().value());
response5.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的过滤字段同样使用source传入参数，与RestAPI相同，使用includes和excludes来标记白名单或黑名单模式，其中includes代表白名单，只返回指定的字段。excludes代表黑名单，不返回指定的字段。

7. 组合查询

// 组合查询
SearchResponse<Test> response6 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .bool(b -> b
                                .must(must -> must
                                        .match(m -> m
                                                .field("age")
                                                .query(30)
                                        )
                                )
                                .must(must -> must
                                        .match(m -> m
                                                .field("sex")
                                                .query("男")
                                        )
                                )
                                .should(should -> should
                                        .match(m -> m
                                                .field("age")
                                                .query(30)
                                        )
                                )
                                .should(should -> should
                                        .match(m -> m
                                                .field("age")
                                                .query(40)
                                        )
                                )
                        )
                )
        , Test.class);
System.out.println(response6.took());
System.out.println(response6.hits().total().value());
response6.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的组合查询，与RestAPI保持一直，使用bool下的must或should来代表必须满足某条件或只需满足某条件。

8. 范围查询

// 范围查询
SearchResponse<Test> response7 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .range(r -> r
                                .field("age")
                                .gte(JsonData.of(30))
                                .lt(JsonData.of(40))
                        )
                )
        , Test.class);
System.out.println(response7.took());
System.out.println(response7.hits().total().value());
response7.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的范围查询，与组合查询不同的是，使用了range而不是bool。field参数代表需要判断的字段，ge、gte、lt、lte分别代表大于、大于等于、小于、小于等于。

9. 模糊查询

// 模糊查询
SearchResponse<Test> response8 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .fuzzy(f -> f
                                .field("name")
                                .value("wangwu")
                                .fuzziness("1"))
                )
        , Test.class);
System.out.println(response8.took());
System.out.println(response8.hits().total().value());
response8.hits().hits().forEach(e -> System.out.println(e.source().toString()));

Elasticsearch Java API Client客户端中的模糊查询，使用fuzzy而不是like，其中field代表需要判断的字段名称，value代表需要模糊查询的关键词，fuzziness代表可以与关键词有误差的字数，可选值为0、1、2这三项。

10. 高亮查询

// 高亮查询
SearchResponse<Test> response9 = client.search(s -> s
                .index("newapi")
                .query(q -> q
                        .term(t -> t
                                .field("name")
                                .value("wangwu")
                        )
                )
                .highlight(h -> h
                        .fields("name", f -> f
                                .preTags("<font color='red'>")
                                .postTags("</font>")
                        )
                )
        , Test.class);
System.out.println(response9.took());
System.out.println(response9.hits().total().value());
response9.hits().hits().forEach(e -> {
    System.out.println(e.source().toString());
    for (Map.Entry<String, List<String>> entry : e.highlight().entrySet()) {
        System.out.println("Key = " + entry.getKey());
        entry.getValue().forEach(System.out::println);
    }
});

Elasticsearch Java API Client客户端中的高亮查询，主要用于给查询出的关键词添加一个标识符，便于前端展示。使用highlight字段，其中fields的key代表需要标记的字段名称，preTags代表需要添加标记的前缀，postTags代表需要添加标记的后缀。同时响应的获取方式也有所改变，具体可以参照上述代码。

11. 聚合查询

// 聚合查询
SearchResponse<Test> response10 = client.search(s -> s
                .index("newapi")
                .aggregations("maxAge", a -> a
                        .max(m -> m
                                .field("age")
                        )
                )
        , Test.class);
System.out.println(response10.took());
System.out.println(response10.hits().total().value());
response10.hits().hits().forEach(e -> {
    System.out.println(e.source().toString());
});
for (Map.Entry<String, Aggregate> entry : response10.aggregations().entrySet()) {
    System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue().max().value());
}

Elasticsearch Java API Client客户端中的聚合查询，主要用于数据的统计，这里演示一下获取最大值。首先使用的是aggregations方法，aggregations方法的key可以自行起名，max代表最大值，可以参照api获取更多的查询方式，这里只演示下max方法，其他方法与其类似。field代表需要获取最大值的字段名称。响应的获取方式也有所不同，需要拿到响应中的aggregations参数，我这里直接进行Map循环以获取统计出的最大值数据。

12. 分组查询

// 分组查询
SearchResponse<Test> response11 = client.search(s -> s
                .index("newapi")
                .size(100)
                .aggregations("ageGroup", a -> a
                        .terms(t -> t
                                .field("age")
                        )
                )
        , Test.class);
System.out.println(response11.took());
System.out.println(response11.hits().total().value());
response11.hits().hits().forEach(e -> {
    System.out.println(e.source().toString());
});
Aggregate aggregate = response11.aggregations().get("ageGroup");
LongTermsAggregate lterms = aggregate.lterms();
Buckets<LongTermsBucket> buckets = lterms.buckets();
for (LongTermsBucket b : buckets.array()) {
    System.out.println(b.key() + " : " + b.docCount());
}

Elasticsearch Java API Client客户端中的分组查询，也是属于聚合查询的一部分，所以同样使用aggregations方法，并使用terms方法来代表分组查询，field传入需要分组的字段，最后通过响应中的aggregations参数来获取，这里需要根据数据的类型来获取最后的分组结果，我这里因为统计的是数字类型，所以使用LongTermsAggregate来获取结果，最后打印出docCount属性即可。