【Hadoop】熟悉常用的HBase操作（Java实现）

短命女 2023-07-17 15:31 7阅读 0赞

> # 一、实验平台 #
> 
>  *  操作系统：Linux(deepin)
>  *  Hadoop版本：2.7.7
>  *  HBase版本：1.2.6
>  *  Java IDE：Eclipse

> # 二、实验内容 #
> 
> ### 1.使用Hadoop提供的Java编程共同实现以下指定功能： ###
> 
> 1.  列出HBase所有的表的相关信息，例如表名；
> 2.  在终端打印出指定的表的所有记录数据；
> 3.  向已经创建好的表添加和删除指定的列族或列；
> 4.  清空指定的表的所有记录数据；
> 5.  统计表的行数
> 
> ### 2.使用Java编程实现以下函数和功能： ###
> 
> 1.  **createTable(String tableName, String\[\] fields)**  
>     创建表，参数tableName为表的名称，字符串数组fields为存储记录各个域名称的数组。要求当HBase已经存在名为tableName的表的时候，先删除原有的表，然后再创建新的表。
> 2.  **addRecord(String tableName, String row, String\[\] fields, String\[\] values)**  
>     向表tableName、行row（用S\_Name表示）和字符串数组files指定的单元格中添加对应的数据values。其中fields中每个元素如果对应的列族下还有相应的列限定符的话，用“columnFamily:column”表示。例如，同时向“Math”、“Computer Science”、“English”三列添加成绩时，字符串数组fields为\{“Score:Math”,”Score；Computer Science”,”Score:English”\}，数组values存储这三门课的成绩。
> 3.  **scanColumn(String tableName, String column)**  
>     浏览表tableName某一列的数据，如果某一行记录中该列数据不存在，则返回null。要求当参数column为某一列族名称时，如果底下有若干个列限定符，则要列出每个列限定符代表的列的数据；当参数column为某一列具体名称（例如“Score:Math”）时，只需要列出该列的数据。
> 4.  **modifyData(String tableName, String row, String column, String val)**  
>     修改表tableName，行row（可以用学生姓名S\_Name表示），列column指定的单元格的数据。
> 5.  **deleteRow(String tableName, String row)**  
>     删除表tableName中row指定的行的记录。

import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.Cell;
    import org.apache.hadoop.hbase.CellUtil;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.HColumnDescriptor;
    import org.apache.hadoop.hbase.HTableDescriptor;
    import org.apache.hadoop.hbase.MasterNotRunningException;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.ZooKeeperConnectionException;
    import org.apache.hadoop.hbase.client.Admin;
    import org.apache.hadoop.hbase.client.Connection;
    import org.apache.hadoop.hbase.client.ConnectionFactory;
    import org.apache.hadoop.hbase.client.Delete;
    import org.apache.hadoop.hbase.client.Get;
    import org.apache.hadoop.hbase.client.HBaseAdmin;
    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.apache.hadoop.hbase.client.Table;
    import org.apache.hadoop.hbase.util.Bytes;
    
    import java.util.Scanner;
    
    public class HbaseTask1 {
    	// 对配置信息管理的一个类
    	public static Configuration configuration;
    	// 对连接进行管理的一个类
    	public static Connection connection;
    	// 对数据库进行管理的一个类，用于对表的增删改查
    	public static Admin admin;
    	
    	// 建立链接
    	public static void init() {
    		configuration = HBaseConfiguration.create();
    		configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");
    		configuration.set("hbase.zookeeper.property.clientPort", "2181");
    		configuration.set("hbase.zookeeper.property.quorum", "localhost");
    		
    		try {
    			// 获取连接池
    			connection = ConnectionFactory.createConnection(configuration);
    			// 使用连接对象获取Admin对象
    			admin = connection.getAdmin();
    		} catch(IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	// 关闭连接
    	public static void close() {
    		try {
    			if(admin != null) {
    				admin.close();
    			}
    			if(connection != null) {
    				connection.close();
    			}
    		}catch(IOException e) {
    			e.printStackTrace();
    		}
    	}
    	/*
    	 * 1.1 列出HBase所有的表的相关信息，例如表名；
    	 */
    	public static void list() throws IOException {
    		init();
    		System.out.println("打印所有表的相关信息：");
    		HTableDescriptor htds[] = admin.listTables();
    		for(HTableDescriptor htd : htds) {
    			System.out.println(htd.getNameAsString());
    		}
    		close();
    	}
    	/*
    	 * 1.2 在终端打印出指定的表的所有记录数据；
    	 */
    	public static void scan(String tablename) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tablename));
    		Scan s = new Scan();
    		ResultScanner scanner = table.getScanner(s);
    		System.out.println("打印表"+tablename+"的所有信息：");
    		for (Result result : scanner) {
    			Cell[] cells = result.rawCells();
    			for (Cell cell : cells) {
    				System.out.println("rowKey: " + new String(CellUtil.cloneRow(cell)));
    				System.out.println("colFamily: " + new String(CellUtil.cloneFamily(cell)));
    				System.out.println("clomn: " + new String(CellUtil.cloneQualifier(cell)));
    				System.out.println("timesTamp: " + cell.getTimestamp());
    				System.out.println("value: " + new String(CellUtil.cloneValue(cell)));
    			}
    		}
    		close();
    	}
    	
    	/*
    	 * 1.3 向已经创建好的表添加和删除指定的列族或列;
    	 */
    	public static void put(String tablename, String rowKey, String colFamily, String colmn, String value) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tablename));
    		Put p = new Put(rowKey.getBytes());
    		p.addColumn(colFamily.getBytes(), colmn.getBytes(), value.getBytes());
    		table.put(p);
    		table.close();
    		close();
    	}
    	
    	public static void delete(String tablename, String rowKey, String colFamily, String colmn, String value, int flag) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tablename));
    		Delete del = new Delete(rowKey.getBytes());
    		
    		if (flag == 1) {
    			// flag为1，删除指定列族的所有数据
    			del.addFamily(colFamily.getBytes());
    			table.delete(del);
    		} else {
    			// flag为2，删除指定列的数据
    			del.addColumn(colFamily.getBytes(), colmn.getBytes());
    			table.delete(del);
    		}
    		
    		table.close();
    		close();
    	}
    	/*
    	 * 1.4 清空指定的表的所有记录数据；
    	 */
    	public static void truncate(String tableName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
    		init();
    		// 用admin2保存表的架构，重建时用
    		HBaseAdmin admin2 = new HBaseAdmin(configuration);
    		HTableDescriptor htd = admin2.getTableDescriptor(tableName.getBytes());
    		
    		TableName tablename = TableName.valueOf(tableName);
    		// 关闭表
    		admin.disableTable(tablename);
    		// 删除表
    		admin.deleteTable(tablename);
    		// 重建表
    		admin.createTable(htd);
    		close();
    	}
    	/*
    	 * 1.5 统计表的行数。
    	 */
    	public static void count(String tablename) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tablename));
    		Scan s = new Scan();
    		ResultScanner scanner = table.getScanner(s);
    		int num = 0;
    		for (Result result : scanner) {
    			if (result != null) {
    				num ++;				
    			}
    		}
    		System.out.println("行数为："+num);
    		close();
    	}
    	/*
    	 * 2.1 创建表，参数tableName为表的名称，字符串数组fields为存储记录各个域名称的数组。
    	 * 要求当HBase已经存在名为tableName的表的时候，先删除原有的表，然后再创建新的表。
    	 */
    	public static void createTable(String tableName, String[] fields) throws IOException {
    		init();
    		TableName tablename = TableName.valueOf(tableName);
    		if (admin.tableExists(tablename)) {
    			System.out.println("该表已存在，删除后重建。");
    			admin.disableTable(tablename);
    			admin.deleteTable(tablename);
    		}
    
    		HTableDescriptor htd = new HTableDescriptor(tablename);
    		for (String str : fields) {
    			HColumnDescriptor hcd = new HColumnDescriptor(str);
    			htd.addFamily(hcd);
    		}
    		admin.createTable(htd);
    		close();
    	}
    
    	/*
    	 * 2.2 向表tableName、行row（用S_Name表示）和字符串数组files指定的单元格中添加对应的数据values。
    	 * 其中fields中每个元素如果对应的列族下还有相应的列限定符的话，用“columnFamily:column”表示。
    	 * 例如，同时向“Math”、“Computer Science”、“English”三列添加成绩时，
    	 * 字符串数组fields为{“Score:Math”,”Score:Computer Science”,”Score:English”}，
    	 * 数组values存储这三门课的成绩.
    	 */
    	public static void addRecord(String tableName, String row, String[] fields, String[] values) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tableName));
    		for (int i = 0; i < fields.length; i++) {
    			Put p = new Put(row.getBytes());
    			String[] cols = fields[i].split(":");
    			if (cols.length == 1) {
    				p.addColumn(cols[0].getBytes(), "".getBytes(), values[i].getBytes());
    			} else {
    				p.addColumn(cols[0].getBytes(), cols[1].getBytes(), values[i].getBytes());
    			}
    			table.put(p);
    		}
    		table.close();
    		close();
    	}
    
    	/*
    	 * 2.3 浏览表tableName某一列的数据，如果某一行记录中该列数据不存在，则返回null。
    	 * 要求当参数column为某一列族名称时，如果底下有若干个列限定符，则要列出每个列限定符代表的列的数据；
    	 * 当参数column为某一列具体名称（例如“Score:Math”）时，只需要列出该列的数据。
    	 */
    	public static void scanColumn(String tableName, String column) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tableName));
    		Scan scan = new Scan();
    		String[] cols = column.split(":");
    		if (cols.length == 1) {
    			scan.addFamily(Bytes.toBytes(column));
    		} else {
    			scan.addColumn(Bytes.toBytes(cols[0]), Bytes.toBytes(cols[1]));
    		}
    		ResultScanner scanner = table.getScanner(scan);
    		for (Result result : scanner) {
    			Cell[] cells = result.rawCells();
    			for (Cell cell : cells) {
    				System.out.println("rowKey: " + new String(CellUtil.cloneRow(cell)));
    				System.out.println("colFamily: " + new String(CellUtil.cloneFamily(cell)));
    				System.out.println("clomn: " + new String(CellUtil.cloneQualifier(cell)));
    				System.out.println("timesTamp: " + cell.getTimestamp());
    				System.out.println("value: " + new String(CellUtil.cloneValue(cell)));
    			}
    		}
    		table.close();
    		close();
    	}
    
    	/*
    	 * 2.4 修改表tableName，行row（可以用学生姓名S_Name表示），列column指定的单元格的数据。
    	 */
    
    	public static void modifyData(String tableName, String row, String column, String val) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tableName));
    		Put p = new Put(row.getBytes());
    		String[] cols = column.split(":");
    		if (cols.length == 1) {
    			p.addColumn(column.getBytes(), "".getBytes(), val.getBytes());
    		} else {
    			p.addColumn(cols[0].getBytes(), cols[1].getBytes(), val.getBytes());
    		}
    		table.put(p);
    		table.close();
    		close();
    	}
    
    	/*
    	 * 2.5 删除表tableName中row指定的行的记录。
    	 */
    	public static void deleteRow(String tableName, String row) throws IOException {
    		init();
    		Table table = connection.getTable(TableName.valueOf(tableName));
    		Delete delete = new Delete(row.getBytes());
    		table.delete(delete);
    		table.close();
    		close();
    	}
    
    }