一、实验目的
1. 理解分布式数据存储系统HBase的架构和工作原理。
2. 掌握HBase表的设计原则,能够根据实际业务需求设计合理的表结构。
3. 学习使用HBase Java API进行数据的插入、查询和管理。
4. 实践分布式数据存储系统在大数据环境下的应用,提升实际操作能力和问题解决能力。
二、安装配置HBase集群
1、安装并配置一个HBase集群,确保集群中的各个组件正常运行。
2、确保集群中的主节点(Master)和多个从节点(RegionServers)都可以正常通信。
三、设计HBase表结构
根据钞票交易数据的业务需求,设计合适的HBase表结构,考虑如何存储和检索钞票交易数据。
HBase表设计:
表名:currency_transactions
行键:<冠字号>#<交易时间>(确保唯一性,按时间查询)
列族:
info:存储钞票基本信息,如面额、交易金额。
transaction:存储交易相关信息,如交易时间、交易地点、类型。
meta:其他信息。
四、插入部分钞票交易数据并探索数据特征
1、将一部分钞票交易数据插入到设计的HBase表中。
(1)创建表:create 'currency_transactions', 'info', 'transaction', 'meta'
插入数据:
put 'currency_transactions', '123456ABC#20241201', 'info:denomination', '100'
put 'currency_transactions', '123456ABC#20241201', 'info:amount', '100'
put 'currency_transactions', '123456ABC#20241201', 'transaction:time', '2024-12-01 10:00'
put 'currency_transactions', '123456ABC#20241201', 'transaction:location', 'Beijing'
put 'currency_transactions', '123456ABC#20241201', 'transaction:type', 'Deposit'
put 'currency_transactions', '123456ABC#20241201', 'meta:notes', 'First deposit'
使用基础的HBase查询语句,探索钞票交易数据的结构和特征,确保数据可以正确存储和访问。
(1)查看表中所有数据:scan 'currency_transactions'
(2)查询特定行键的数据:
get 'currency_transactions', '123456ABC#20241201'
(3)查询特定列的数据:get 'currency_transactions', '123456ABC#20241201', 'transaction:location'
五、使用HBase Java API进行操作
1、创建Maven项目并添加相关依赖
(1)在IDEA上面新建Maven项目,下载与虚拟机中HBase一致的jdk版本1.8
(2)新建项目
(3)编辑pom.xml文件,内容如下:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>javaapi</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><!-- HBase Client --><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.2.5</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency></dependencies>
</project>
log4j.properties:
# Set root logger level and appender
log4j.rootLogger=INFO, console
# Console appender
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
2、HBase Java API的基本操作之表的创建
(1)代码部分:
package com.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;import java.io.IOException;public class HBaseTableCreator {public static void main(String[] args) throws IOException {Configuration config = HBaseConfiguration.create();config.addResource(new Path("hbase-site.xml"));config.set("hbase.zookeeper.quorum", "192.168.125.101");config.set("hbase.zookeeper.property.clientPort", "2181");try (Connection connection = ConnectionFactory.createConnection(config);Admin admin = connection.getAdmin()) {TableName tableName = TableName.valueOf("currency_transactions");// 使用 HBase 1.x API 定义列族HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);tableDescriptor.addFamily(new HColumnDescriptor("info"));tableDescriptor.addFamily(new HColumnDescriptor("transaction"));tableDescriptor.addFamily(new HColumnDescriptor("meta"));// 检查表是否存在if (admin.tableExists(tableName)) {System.out.println("Table already exists. Deleting and recreating...");admin.disableTable(tableName);admin.deleteTable(tableName);}// 创建表admin.createTable(tableDescriptor);System.out.println("Table created successfully.");}}
}
(2)运行结果:
3、HBase Java API的基本操作——数据的插入
(1)代码部分:
package com.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;import java.io.IOException;public class HBaseDataInserter {public static void main(String[] args) throws IOException {// 配置 HBase 连接Configuration config = HBaseConfiguration.create();config.addResource(new Path("hbase-site.xml"));config.set("hbase.zookeeper.quorum", "192.168.125.101");config.set("hbase.zookeeper.property.clientPort", "2181");// 定义表名String tableName = "currency_transactions";try (Connection connection = ConnectionFactory.createConnection(config)) {// 插入多行数据addRows(connection, tableName);}}public static void addRows(Connection connection, String tableName) throws IOException {try (Table table = connection.getTable(TableName.valueOf(tableName))) {// 创建 Put 对象,插入行1 (Alice)Put put1 = new Put("row1".getBytes());put1.addColumn("info".getBytes(), "name".getBytes(), "Alice".getBytes());put1.addColumn("info".getBytes(), "age".getBytes(), "30".getBytes());put1.addColumn("transaction".getBytes(), "amount".getBytes(), "1000".getBytes());put1.addColumn("transaction".getBytes(), "currency".getBytes(), "USD".getBytes());// 创建 Put 对象,插入行2 (Bob)Put put2 = new Put("row2".getBytes());put2.addColumn("info".getBytes(), "name".getBytes(), "Bob".getBytes());put2.addColumn("info".getBytes(), "age".getBytes(), "40".getBytes());put2.addColumn("transaction".getBytes(), "amount".getBytes(), "500".getBytes());put2.addColumn("transaction".getBytes(), "currency".getBytes(), "EUR".getBytes());// 批量插入table.put(put1);table.put(put2);System.out.println("Rows added successfully.");}}
}
(2)运行结果:
4、根据冠字号(行键)查询单行数据
(1)实现代码
package com.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;import java.io.IOException;public class HBaseDataRetriever {public static void main(String[] args) throws IOException {// 配置 HBase 连接Configuration config = HBaseConfiguration.create();config.addResource(new Path("hbase-site.xml"));config.set("hbase.zookeeper.quorum", "192.168.125.101");config.set("hbase.zookeeper.property.clientPort", "2181");// 定义表名和行键(冠字号)String tableName = "currency_transactions";String rowKey = "row1"; // 冠字号对应的行键try (Connection connection = ConnectionFactory.createConnection(config)) {// 查询单行数据retrieveRowByKey(connection, tableName, rowKey);}}public static void retrieveRowByKey(Connection connection, String tableName, String rowKey) throws IOException {try (Table table = connection.getTable(TableName.valueOf(tableName))) {// 创建 Get 对象Get get = new Get(rowKey.getBytes());// 指定需要的列族和列(可选)get.addFamily("info".getBytes()); // 获取 "info" 列族的所有列get.addColumn("transaction".getBytes(), "amount".getBytes()); // 获取特定列// 获取结果Result result = table.get(get);// 遍历结果System.out.println("Row Key: " + rowKey);result.listCells().forEach(cell -> {String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());System.out.println(family + ":" + qualifier + " = " + value);});}}
}
(2)运行结果
5、批量检索所有数据(扫描)
(1)实现代码
package com.example;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;import java.io.IOException;public class HBaseDataScanner {public static void main(String[] args) throws IOException {// 配置 HBase 连接Configuration config = HBaseConfiguration.create();config.addResource(new Path("hbase-site.xml"));config.set("hbase.zookeeper.quorum", "192.168.125.101");config.set("hbase.zookeeper.property.clientPort", "2181");// 定义表名String tableName = "currency_transactions";try (Connection connection = ConnectionFactory.createConnection(config)) {// 扫描表数据scanTableData(connection, tableName);}}public static void scanTableData(Connection connection, String tableName) throws IOException {try (Table table = connection.getTable(TableName.valueOf(tableName))) {// 创建 Scan 对象Scan scan = new Scan();// 指定列族或列(可选)scan.addFamily("info".getBytes()); // 扫描 "info" 列族scan.addColumn("transaction".getBytes(), "currency".getBytes()); // 扫描特定列// 获取结果ResultScanner scanner = table.getScanner(scan);// 遍历结果for (Result result : scanner) {String rowKey = new String(result.getRow());System.out.println("Row Key: " + rowKey);result.listCells().forEach(cell -> {String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());System.out.println(family + ":" + qualifier + " = " + value);});System.out.println("------------");}}}
}
(2)运行结果
6、根据列值(如冠字号、金额等)过滤数据
(1)实现代码
package com.example;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.hbase.HBaseConfiguration;import org.apache.hadoop.hbase.TableName;import org.apache.hadoop.hbase.client.*;import org.apache.hadoop.hbase.filter.*;import java.io.IOException;public class HBaseFilteredScan {public static void main(String[] args) throws IOException {// 配置 HBase 连接Configuration config = HBaseConfiguration.create();config.addResource(new Path("hbase-site.xml"));config.set("hbase.zookeeper.quorum", "192.168.125.101");config.set("hbase.zookeeper.property.clientPort", "2181");// 定义表名String tableName = "currency_transactions";try (Connection connection = ConnectionFactory.createConnection(config)) {// 使用过滤器检索数据scanTableWithFilter(connection, tableName, "transaction", "currency", "USD");}}public static void scanTableWithFilter(Connection connection, String tableName, String columnFamily, String columnQualifier, String valueToFilter) throws IOException {try (Table table = connection.getTable(TableName.valueOf(tableName))) {// 创建 Scan 对象Scan scan = new Scan();// 添加列值过滤器SingleColumnValueFilter filter = new SingleColumnValueFilter(columnFamily.getBytes(), // 列族columnQualifier.getBytes(), // 列名CompareFilter.CompareOp.EQUAL, // 比较操作valueToFilter.getBytes()); // 目标值// 设置过滤器scan.setFilter(filter);// 获取结果ResultScanner scanner = table.getScanner(scan);// 遍历结果for (Result result : scanner) {String rowKey = new String(result.getRow());System.out.println("Row Key: " + rowKey);result.listCells().forEach(cell -> {String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());System.out.println(family + ":" + qualifier + " = " + value);});System.out.println("------------");}}}
}
(2)运行结果