MRS基础组件之HBase与Hive开发应用_tyt2023

标签：xxx hadoop Hive MRS import apache org HBase hbase

MRS基础组件之HBase与Hive开发应用

本实验基于MRS环境，介绍如何利用HBase与Hive来进行相关操作。其中，HBase主要介绍包括如何利用Java API创建数据表、写入数据、查看数据以及删除数据；而Hive则通过介绍UDF、UDTF和UDAF等自定义函数的基本操作，讲解如何进行自定义数据处理和清洗作业。

购买相关云服务

购买弹性公网IP EIP

购买 “产品->大数据->MapReduce服务”

选择“自定义购买”

区域：华北-北京四

集群名称：mrs-bigdata

版本类型：普通版

集群版本：MRS 3.1.0 WXL

集群类型：自定义

勾选组件：Hadoop/HBase/Hive/Zookeeper/Ranger

确认无误后点击“下一步”。

接下来配置硬件。

计费模式：按需计费。

可用区：任意均可

虚拟私有云：vpc-bigdata

子网：subnet-bigdata

安全组：sg-bigdata

弹性公网IP：选择下拉框中的任意IP即可

开启“拓扑调整”，勾选如下图位置所示的“DN, NM, RS”。此操作表示在Master3节点分别部署DataNode, NodeManager, RegionServer以解决如上警告。

安装jdk环境

wget https://sandbox-expriment-files.obs.cn-north-1.myhuaweicloud.com/hccdp/HCCDP/jdk-8u341-linux-x64.tar.gz

HBase实验

HBase是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，是行业中最常用的NoSQL数据库。掌握HBase的使用，可加深学员对HBase的理解，为综合应用大数据打下坚实的基础。本实验主要讲述了对HBase非关系型数据库进行的JavaAPI操作。该实验旨在指导用户掌握如何使用Java API来调用HBase，以便在HBase数据库中创建表、查看表、删除表，并且也可以向数据库中的表里面的数据进行添加数据、查看数据、删除数据等操作。

查看Zookeeper集群IP地址

将Zookeeper三个quorumpeer实例的“管理IP”地址记录下来

创建Maven项目

按如下配置项目信息：

① GroupId：com.huawei

② Artifactid：HBaseAPI

③ Version：0.0.1-SNAPSHOT

④ Packaging：jar

修改JDK路径

右上角选择Window标签，在下拉菜单最后一栏中找到Preferences。

回到初始界面，看到项目名称下有一个类似JRE System Library [J2SE-1.5]的标签。右键点击该标签，选择Build Path->Configure Build Path:

在新窗口点击Add Library

选择JRE System Library，点击Next。

将之前的J2SE-1.5直接删除。选择该模块，在右边找到Remove：

4配置POM文件

<?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.huawei</groupId>
<artifactId>HbaseAPI</artifactId>
<version>1.0-SNAPSHOT</version>

<repositories>
<repository>
<id>huaweicloud2</id>
<name>huaweicloud2</name>
<url>https://mirrors.huaweicloud.com/repository/maven/</url>
</repository>
<repository>
<id>huaweicloud1</id>
<name>huaweicloud1</name>
<url>https://repo.huaweicloud.com/repository/maven/huaweicloudsdk/</url>
</repository>
</repositories>

<build>
<finalName>HbaseAPI</finalName>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>assembly</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

配置好pom文件后，使用“ctrl+s”进行刷新，下载jar包需要一定的时间等待，视网络情况而定预计约需2-3分钟。

创建表

在src>main>java目录下创建包com.huawei.hbase.tables。右键点击src/main/java，选择Package。编辑Name为：com.huawei.hbase.tables,完成后点击Finish。

在这个包下面创建名为”CreateTable”的Java class，来实现在HBase数据库中创建users表，其中列族为“f”。

右键点击创建好的com.huawei.hbase.tables包，选择New -> Class。

输入类名CreateTable，并点击Finish。注意该类是在com.huawei.hbase.tables包下创建，如果不是请重新创建。

此时我们得到了一个空的CreateTable类。下面需要在该类中进行业务代码开发。具体代码如下（输入完成后使用“ctrl+s”进行保存）。

注意：代码中Zookeeper的IP地址需要根据实际情况进行修改,选择集群中的Zookeeper任一服务地址，端口保持不变。

package com.huawei.hbase.tables;

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.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;

/**
* 创建Hbase表
*/

public class CreateTable {

public static void main(String[] args) throws IOException {

//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");

conf.set("hbase.zookeeper.property.clientPort", "2181");

Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();

TableName tableName = TableName.valueOf("users");

if (!admin.tableExists(tableName)){
//创建表描述器
HTableDescriptor htd = new HTableDescriptor(tableName);
htd.addFamily(new HColumnDescriptor("f"));
admin.createTable(htd);
System.out.println(tableName+"表创建成功");
}else {
System.out.println(tableName+"表已经存在");
}
}
}

查看表结构

参考上一步骤的相关操作，在com.huawei.hbase.tables包下创建名为DescTable的Java类，并点击Finish，用于查看users表的结构形式。

package com.huawei.hbase.tables;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;

/**
* 查看表结构
*/
public class DescTable {
public static void main(String[] args) throws IOException {

//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");

Connection connection = ConnectionFactory.createConnection(conf);
Admin hBaseAdmin = connection.getAdmin();

TableName tableName = TableName.valueOf("users");
if(hBaseAdmin.tableExists(tableName)) {
HTableDescriptor htd = hBaseAdmin.getTableDescriptor(tableName);
System.out.println("查看"+tableName+"表结构");
System.out.println(htd);
}else {
System.out.println(tableName+"表不存在");
}
}
}

删除表

参考之前的相关操作，在com.huawei.hbase.tables包下创建名为DeleteTable的Java类，用于删除HBase数据库中的users表（其中“xxx.xxx.xxx.xxx” 为集群中的Zookeeper任一服务地址，端口保持不变。）。

package com.huawei.hbase.tables;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;

/**
* 删除表
*/
public class DeleteTable {
public static void main(String[] args) throws IOException {

//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");

Connection connection = ConnectionFactory.createConnection(conf);
Admin hBaseAdmin = connection.getAdmin();

TableName tableName = TableName.valueOf("users");

if (hBaseAdmin.tableExists(tableName)) {
//判断表的状态
if(hBaseAdmin.isTableAvailable(tableName)) {
hBaseAdmin.disableTable(tableName);
}
hBaseAdmin.deleteTable(tableName);
System.out.println("删除表"+tableName+"成功");
}else {
System.out.println(tableName+"表不存在");
}
}
}

添加数据

参考之前的相关操作，新建一个名为com.huawei.hbase.data的Package

在该Package下新建一个名为PutData的Java Class。

相关代码如下（其中“xxx.xxx.xxx.xxx” 为集群中的Zookeeper任一服务地址，端口保持不变）：

package com.huawei.hbase.data;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class PutData {
public static void main(String[] args) throws IOException {
//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");

Connection connection = ConnectionFactory.createConnection(conf);

Table hTable = connection.getTable(TableName.valueOf("users"));

//插入一条
Put put= new Put(Bytes.toBytes("row1"));
put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"),Bytes.toBytes("1"));
put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("name"),Bytes.toBytes("zjhw"));
put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("age"),Bytes.toBytes(27));
put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("phone"),Bytes.toBytes("18238485602"));
put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("emil"),Bytes.toBytes("[email protected]"));
hTable.put(put);
//插入多个
Put put1= new Put(Bytes.toBytes("row2"));
put1.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"),Bytes.toBytes("2"));
put1.addColumn(Bytes.toBytes("f"),Bytes.toBytes("name"),Bytes.toBytes("zhehua"));

Put put2= new Put(Bytes.toBytes("row3"));
put2.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"),Bytes.toBytes("3"));
put2.addColumn(Bytes.toBytes("f"),Bytes.toBytes("name"),Bytes.toBytes("zhejiang"));

Put put3= new Put(Bytes.toBytes("row4"));
put3.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"),Bytes.toBytes("4"));
put3.addColumn(Bytes.toBytes("f"),Bytes.toBytes("name"),Bytes.toBytes("bigdata"));

List<Put> list = new ArrayList<Put>();
list.add(put1);
list.add(put2);
list.add(put3);
hTable.put(list);

//检测put,条件成功就插入,要求RowKey是一样的
Put put4 = new Put(Bytes.toBytes("row5"));
put4.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"),Bytes.toBytes("5"));
hTable.checkAndPut(Bytes.toBytes("row5"),Bytes.toBytes("f"),Bytes.toBytes("id"),null,put4);

System.out.println("插入成功");
}
}

使用Get命令查询数据

在com.huawei.hbase.data 包下新建一个名为GetData的Java Class，用于查询RowKey为row1的数据（其中“xxx.xxx.xxx.xxx” 为集群中的Zookeeper任一服务地址，端口保持不变）。

package com.huawei.hbase.data;

import java.io.IOException;

public class GetData {
public static void main(String[] args) throws IOException {
//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");
Connection connection = ConnectionFactory.createConnection(conf);

Table hTable = connection.getTable(TableName.valueOf("users"));

Get get = new Get(Bytes.toBytes("row1"));
Result result = hTable.get(get);
byte[] family = Bytes.toBytes("f");
byte[] buf = result.getValue(family,Bytes.toBytes("id"));
System.out.println("id="+Bytes.toString(buf));

buf = result.getValue(family,Bytes.toBytes("age"));
System.out.println("age="+Bytes.toInt(buf));

buf = result.getValue(family,Bytes.toBytes("name"));
System.out.println("name="+Bytes.toString(buf));

buf = result.getRow();
System.out.println("rowkey="+Bytes.toString(buf));
}
}

使用Scan命令扫描数据

在com.huawei.hbase.data 包下新建一个名为ScanData的Java Class，用于查询RowKey范围为row1到row5中id和name两列的信息（其中“xxx.xxx.xxx.xxx” 为集群中的Zookeeper任一服务地址，端口保持不变）。

package com.huawei.hbase.data;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.MultipleColumnPrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;

public class ScanData {
public static void main(String[] args) throws IOException {
//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");
Connection connection = ConnectionFactory.createConnection(conf);

Table hTable = connection.getTable(TableName.valueOf("users"));

Scan scan = new Scan();
//增加起始rowkey
scan.withStartRow(Bytes.toBytes("row1"));
scan.withStopRow(Bytes.toBytes("row5"));

//增加过滤filter
FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
byte[][] prefixes = new byte[2][];
prefixes[0] = Bytes.toBytes("id");
prefixes[1] = Bytes.toBytes("name");
MultipleColumnPrefixFilter mcpf = new MultipleColumnPrefixFilter(prefixes);
list.addFilter(mcpf);
scan.setFilter(list);

ResultScanner rs = hTable.getScanner(scan);
Iterator<Result> iter = rs.iterator();
while (iter.hasNext()){
Result result = iter.next();
printResult(result);
}
}

/*
打印Result对象
*/
static void printResult(Result result){
System.out.println("***********"+Bytes.toString(result.getRow()));
NavigableMap<byte[],NavigableMap<byte[], NavigableMap<Long,byte[]>>> map = result.getMap();
for(Map.Entry<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> entry: map.entrySet()){
String family = Bytes.toString(entry.getKey());
for(Map.Entry<byte[],NavigableMap<Long,byte[]>> columnEntry :entry.getValue().entrySet()){
String column = Bytes.toString(columnEntry.getKey());
String value = "";
if("age".equals(column)){
value=""+Bytes.toInt(columnEntry.getValue().firstEntry().getValue());
}else {
value=""+Bytes.toString(columnEntry.getValue().firstEntry().getValue());
}
System.out.println(family+":"+column+":"+value);
}
}
}
}

删除表中数据

在com.huawei.hbase.data包下新建一个名为DeleteData的Java Class，用来实现删除users表中RowKey为row5的所有信息（其中“xxx.xxx.xxx.xxx” 为集群中的Zookeeper任一服务地址，端口保持不变）

package com.huawei.hbase.data;

import java.io.IOException;

public class DeleteData {
public static void main(String[] args) throws IOException {
//链接hbase
Configuration conf = HBaseConfiguration.create();
//这里的zookeeper地址要改为自己集群的zookeeper地址
conf.set("hbase.zookeeper.quorum",
"xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
conf.set("hbase.zookeeper.property.clientPort", "2181");
Connection connection = ConnectionFactory.createConnection(conf);

Table hTable = connection.getTable(TableName.valueOf("users"));

Delete delete = new Delete(Bytes.toBytes("row5"));
delete.addColumn(Bytes.toBytes("f"),Bytes.toBytes("id"));
//直接删除family,将所有row5的信息全部删除
delete.addFamily(Bytes.toBytes("f"));
hTable.delete(delete);
System.out.println("删除成功");
}
}

2Maven打包并上传

在左侧标签栏，右键项目名称HBaseAPI，选择Run As->Maven Install进行打包。

打包需要一定时间，请耐心等待。

package结束后会出现BUILD SUCCESS，此时我们就可以在相应的目录下查看jar包。

打开桌面上的xfce终端，利用以下命令进入安装目录：

cd ~/eclipse-workspace/HBaseAPI/target/

使用ls命令即可看到两个jar包。

利用scp命令将HbaseAPI-jar-with-dependencies.jar上传到MRS集群中：

scp ~/eclipse-workspace/HBaseAPI/target/HbaseAPI-jar-with-dependencies.jar [email protected]:/root

运行并验证

（1）使用ssh命令远程登录到MRS集群中。

使用ls命令查看HbaseAPI-jar-with-dependencies.jar是否存在。

（2）使用如下命令运行jar包，在HBase数据库中创建users表。

  yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.tables.CreateTable

使用如下命令运行jar包，查看users表结构。

 yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.tables.DescTable

使用如下命令，向users表中插入数据。

  yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.data.PutData

使用Get查询users表中行键为的row1的数据。

yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.data.GetData

使用Scan扫描users表中行键为row1至row4这4行数据中列族名为f，列名分别为id和name两列的信息。

  yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.data.ScanData

删除users表中行键为row5的数据。

 yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.data.DeleteData

删除users表。

yarn jar HbaseAPI-jar-with-dependencies.jar com.huawei.hbase.tables.DeleteTable Hive自定义函数开发实验 Hive是Hadoop生态系统乃至整个大数据应用开发领域十分重要的数据仓库工具，在数据挖掘、数据汇总、统计分析等领域有重要作用，可以查询和管理PB级别的分布式数据。除了基本的HiveQL命令之外，Hive还提供了非常高的灵活性，允许我们通过Java进行自定义函数的编写及使用。本实验就将介绍如何在Hive中开发UDF、UDTF和UDAF这三大类自定义函数，帮助大家更好地掌握相关开发与应用。

创建Eclipse Maven项目

点击左上角的File标签，选择 ”New” > ”project” 按钮，新建一个项目。

按如下配置项目信息：

① GroupId：com.huawei

② Artifactid：HiveFunction

③ Version：0.0.1-SNAPSHOT

④ Packaging：jar

修改JDK路径

右上角选择Window标签，在下拉菜单最后一栏中找到Preferences。

回到初始界面，看到项目名称下有一个类似JRE System Library [J2SE-1.5]的标签。

右键点击该标签，选择Build Path->Configure Build Path:

将之前的J2SE-1.5直接删除。选择该模块，在右边找到Remove：

配置POM文件

<groupId>com.huawei</groupId>
<artifactId>HiveFunction</artifactId>
<version>1.0-SNAPSHOT</version>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>

<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.8.3</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>2.3.2</version>
</dependency>
</dependencies>

<build>
<finalName>HiveFunction</finalName>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>assembly</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

配置好pom文件后，刷新maven配置，下载实验所需要的jar包，下载jar包需要一定的时间等待。

4创建自定义UDF函数

首先在src/main/java下创建一个名为com.huawei.hiveFunctions的Package。

创建大小写转换自定义函数，自定义UDF，要求继承UDF，并重载实现evaluate方法，一个输入对应一个输出。在com.huawei.hiveFunctions下创建一个名为UDFLowerOrUpperCase的Java类

package com.huawei.hiveFunctions;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

/**
\* 创建大小写转换自定义函数
*/

public class UDFLowerOrUpperCase extends UDF {
//转换为小写
public Text evaluate(Text t){
//默认是进行小写转换
return this.evaluate(t,"lower");
}

/*
对参数进行大小写转换
如果该值为lower，则进行小写转换，如果改值为upper则进行大写转换，其他情况不进行转换
*/
public Text evaluate(Text t,String lowerOrUpper){
if(t == null){
return t;
}

if("lower".equals(lowerOrUpper)){
//select lower_upper(studentname,'lower') from students;
return new Text(t.toString().toLowerCase());
}else if("upper".equals(lowerOrUpper)){
//select lower_upper(studentname,'upper') from students;
return new Text(t.toString().toUpperCase());
}

//转换参数错误的情况下，直接返回原本的值
return t;
}
}

创建自定义的UDAF函数

创建自定义的UDAF函数，实现多个输入一个输出来具体实现函数支持整形和浮点型的sum操作，在com.huawei.hiveFunctions下创建类UDAFSumCase。

package com.huawei.hiveFunctions;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFParameterInfo;
import org.apache.hadoop.hive.serde2.io.DoubleWritable;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.AbstractPrimitiveWritableObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.io.LongWritable;

/**
\* 自定义的UDAF函数实现
\* 实现一个自定义的sum函数，要求函数支持整形和浮点型的sum操作
*/

public class UDAFSumCase extends AbstractGenericUDAFResolver {
@Override
public GenericUDAFEvaluator getEvaluator(GenericUDAFParameterInfo info) throws SemanticException {
if (info.isAllColumns()) {
//函数允许*号查询的时候返回一个true
throw new SemanticException("不允许*号查询");
}
//获取函数参数列表
ObjectInspector[] inspectors = info.getParameterObjectInspectors();
if (inspectors.length != 1) {
throw new UDFArgumentException("只支持一个参数进行查询");
}
AbstractPrimitiveWritableObjectInspector apwoi = (AbstractPrimitiveWritableObjectInspector) inspectors[0];
switch (apwoi.getPrimitiveCategory()) {
case BYTE:
case INT:
case SHORT:
case LONG:
//进行整数的sum操作
return new SumLongEvaluator();
case DOUBLE:
case FLOAT:
//进行浮点型的操作
return new SumDoubleEvaluator();
default:
throw new UDFArgumentException("参数异常");
}
}

/**
\* 进行浮点型的sum操作
*/

static class SumDoubleEvaluator extends GenericUDAFEvaluator {
private PrimitiveObjectInspector inputOI;
static class SumDoubleAgg implements AggregationBuffer {
double sum;
boolean empty;
}

@Override
public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
super.init(m, parameters);
this.inputOI = (PrimitiveObjectInspector) parameters[0];
return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;
}

@Override
public AggregationBuffer getNewAggregationBuffer() throws HiveException {
SumDoubleAgg sda = new SumDoubleAgg();
this.reset(sda);
return sda;
}

@Override
public void reset(AggregationBuffer agg) throws HiveException {
SumDoubleAgg sda = (SumDoubleAgg) agg;
sda.empty = true;
sda.sum = 0;
}

@Override
public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
this.merge(agg, parameters[0]);
}

@Override
public Object terminatePartial(AggregationBuffer agg) throws HiveException {
return this.terminate(agg);
}

@Override
public void merge(AggregationBuffer agg, Object partial) throws HiveException {
if (partial != null) {
SumDoubleAgg sda = (SumDoubleAgg) agg;
sda.sum += PrimitiveObjectInspectorUtils.getDouble(partial, inputOI);
sda.empty = false;
}
}

@Override
public Object terminate(AggregationBuffer agg) throws HiveException {
SumDoubleAgg sda = (SumDoubleAgg) agg;
if (sda.empty) {
return null;
}
return new DoubleWritable(sda.sum);
}
}

/**
* 进行整形的sum操作
*/
static class SumLongEvaluator extends GenericUDAFEvaluator {
private PrimitiveObjectInspector inputOI;

@Override
public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
super.init(m, parameters);
if (parameters.length != 1) {
throw new UDFArgumentException("参数异常");
}
inputOI = (PrimitiveObjectInspector) parameters[0];
return PrimitiveObjectInspectorFactory.writableLongObjectInspector;
}

/**
* 自定义类型
*
* @author gerry
*
*/

static class SumLongAgg implements AggregationBuffer {
long sum;
boolean empty;
}

@Override
public AggregationBuffer getNewAggregationBuffer() throws HiveException {
SumLongAgg sla = new SumLongAgg();
this.reset(sla);
return sla;
}

@Override
public void reset(AggregationBuffer agg) throws HiveException {
SumLongAgg sla = (SumLongAgg) agg;
sla.sum = 0;
sla.empty = true;
}

/**
\* 循环处理会调用的方法
*/

@Override
public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
if (parameters.length != 1) {
throw new UDFArgumentException("参数异常");
}
this.merge(agg, parameters[0]);
}

/**
* 部分聚合后的数据输出
*
*/
@Override
public Object terminatePartial(AggregationBuffer agg) throws HiveException {
return this.terminate(agg);
}

/**
* 合并操作
*/

@Override
public void merge(AggregationBuffer agg, Object partial) throws HiveException {
if (partial != null) {
SumLongAgg sla = (SumLongAgg) agg;
sla.sum += PrimitiveObjectInspectorUtils.getLong(partial, inputOI);
sla.empty = false;
}
}

/**
* 全部输出
*/
@Override
public Object terminate(AggregationBuffer agg) throws HiveException {
SumLongAgg sla = (SumLongAgg) agg;
if (sla.empty) {
return null;
}
return new LongWritable(sla.sum);
}
}
}

创建自定义UDTF函数

创建一个自定义函数UDTF，来实现一个输入多次输出的函数，我们提供好一些数据，通过这些数据的清洗，我们可以实现在数据中读取产品ID，产品名称，价格等。

在com.huawei.hiveFunctions下创建类UDTFCase。

package com.huawei.hiveFunctions;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.*;

/**
* 自定义的UDTF函数
* 实现功能：解析爬虫数据，从数据读取产品ID，产品名称，价格等
*/

public class UDTFCase extends GenericUDTF {
@Override
public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
if(argOIs.getAllStructFieldRefs().size() != 1){
throw new UDFArgumentException("参数异常");
}
ArrayList<String> filedName =new ArrayList<String>();
ArrayList<ObjectInspector> filedOIs = new ArrayList<ObjectInspector>();
filedName.add("id");
filedName.add("name");
filedName.add("price");
filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
return ObjectInspectorFactory.getStandardStructObjectInspector(filedName,filedOIs);
}

public void process(Object[] objects) throws HiveException {
if(objects == null || objects.length !=1){
return;
}

//只有一个参数的情况
String line = objects[0].toString();
//解析
Map<String,String> map = transfoerContent2Map(line);
List<String> result = new ArrayList<String>();
result.add(map.get("p_id"));
result.add(map.get("p_name"));
result.add(map.get("price"));
super.forward(result.toArray(new String[0]));
}

public void close() throws HiveException {
//nothing
super.forward(new String[] {"12345689","close","123"});
}

/*
转换字符串为map对象
*/
static Map<String, String> transfoerContent2Map(String content) {
Map<String, String> map = new HashMap<String, String>();
int i = 0;
String key = "";
StringTokenizer tokenizer = new StringTokenizer(content, "({|}|\"|:|,)");
while (tokenizer.hasMoreTokens()) {
if (++i % 2 == 0) {
// 当前的值是value
map.put(key, tokenizer.nextToken());
} else {
// 当前的值是key
key = tokenizer.nextToken();
}
}
return map;
}
}

Maven打包并上传

在左侧标签栏，右键项目名称HiveFunction，选择Run As->Maven Install进行打包

打包需要一定时间，请耐心等待。

package结束后会出现BUILD SUCCESS，此时我们就可以在相应的目录下查看jar包（如果提示BUILD FAILURE打包失败，可以参考后续解决方法直接下载打好的jar包）。

打开桌面上的xfce终端，利用以下命令进入安装目录：

                                cd ~/eclipse-workspace/HiveFunction/target/

使用ls命令即可看到两个jar包。

注：如果HiveFunction-jar-with-dependencies.jar不存在，应该是打包失败了，可以使用下列命令下载已打包成功的文件：

wget https://sandbox-expriment-files.obs.cn-north-1.myhuaweicloud.com/hccdp/HCCDP/HiveFunction-jar-with-dependencies.jar

此时再次利用ls命令，应该就能看到HiveFunction-jar-with-dependencies.jar文件了。

利用scp命令将HiveFunction-jar-with-dependencies.jar上传到MRS集群中：

scp ~/eclipse-workspace/HiveFunction/target/HiveFunction-jar-with-dependencies.jar [email protected]:/root

运行并验证

（1）使用ssh命令远程登录到MRS集群中。同样在xfce终端中，输入以下命令：

ssh [email protected]

使用ls命令查看HiveFunction-jar-with-dependencies.jar文件是否存在。如果不存在，请重新执行上传步骤。

（2）使用Vim编辑器，准备实验所需数据。

                                vim students

按Esc退出insert模式，输入 :wq 然后回车退出vim编辑器。

将students文件和jar包都上传至HDFS：

命令输入完成后，点击两次“enter”

hdfs dfs -mkdir /user/hive_user

hdfs dfs -put students /user/hive_user

hdfs dfs -put HiveFunction-jar-with-dependencies.jar /user/hive_user

（3）利用beeline命令进入Hive命令行环境。

                                beeline

进入Hive环境后即可执行HiveQL的命令了。

创建一张名为students的数据表

create table students(id int comment 'this is student id, is not null', Name string comment 'this is student name') row format delimited fields terminated by ',' stored as textfile;

将HDFS数据导入students表中

load data inpath '/user/hive_user/students' into table students;

查看数据是否导入成功

select * from students;

将HDFS 上的jar包导入Hive

add jar hdfs://hacluster/user/hive_user/HiveFunction-jar-with-dependencies.jar;

利用jar包创建名为lower_upper的Functions来实现大小写之间的转换

create function lower_upper as 'com.huawei.hiveFunctions.UDFLowerOrUpperCase';

调用lower_upper Function，将students表中的name字段全部转为小写

select lower_upper(name) from students;

调用lower_upper Function，将students表中的name字段全部转为大写

select lower_upper(name,'upper') from students;

（4）创建一个名为mysum Function来实现一个自定义的sum函数，要求函数支持整形和浮点型的sum操作

                                create function mysum as 'com.huawei.hiveFunctions.UDAFSumCase';

调用mysum函数

                                select mysum(1) from students;

（5）创建Functions来实现对数据清洗，从数据读取产品ID，产品名称，价格等

此处需要利用Hive读取HBase数据，但MRS 3.1.0版本中并无直接可用的HBase命令行客户端，因此需前往FusionInsight Manager下载客户端并执行安装。

登录FusionInsight Manager，账号名为admin，密码为购买云服务时自定义的密码。

在右上角选择“更多 > 下载客户端”，弹出“下载客户端”信息提示框。

“选择客户端类型”中选择“完整客户端”。

“选择平台类型”中选择x86_64

勾选“仅保存到如下路径”，单击“确定”开始生成客户端文件。

文件生成后默认保存在主管理节点“/tmp/FusionInsight-Client”。

打开一个新的命令行窗口，登录到MRS集群。

                                ssh [email protected]

进入安装包所在目录，例如“/tmp/FusionInsight-Client”。

                                cd /tmp/FusionInsight-Client

执行如下命令解压安装包到本地目录。

                                tar -xvf FusionInsight_Cluster_1_HBase_Client.tar

解压获取的安装文件。

tar -xvf FusionInsight_Cluster_1_HBase_ClientConfig.tar

进入安装包所在目录。

cd /tmp/FusionInsight-Client/FusionInsight_Cluster_1_HBase_ClientConfig

执行下列命令，安装客户端到指定目录（绝对路径），例如安装到“/opt/client”目录。

./install.sh /opt/client

等待客户端安装完成。执行以下命令切换到客户端目录：

cd /opt/client

执行以下命令配置环境变量：

source bigdata_env

输入hbase shell进入hbase命令行：

                                hbase shell

在HBase中创建data数据表并导入数据。

import java.util.Date

#创建表
create 'data','f'

#插入数据
put 'data',Date.new().getTime(),'f:content','{"p_id":"100001","p_name":"<张家界-凤凰-天门山-玻璃栈道飞机5日游>","price":"2141"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100002","p_name":"<丽江-香格里拉-泸沽湖双飞7日游>","price":"4429"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100003","p_name":"<香格里拉-昆大丽3飞6日游>","price":"2479"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100004","p_name":"<桂林-阳朔-古东瀑布-世外桃源双飞5日游>","price":"2389"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100005","p_name":"<海南三亚-无自费5日游>","price":"2389"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100006","p_name":"<成都-九寨沟-黄龙-花湖-红原-九曲双飞7日游>","price":"3729"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100007","p_name":"<海南三亚5日游>","price":"2168"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100008","p_name":"<海南三亚五星0购物6日游>","price":"2916"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100009","p_name":"<厦门双飞4日游>","price":"1388"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100010","p_name":"<绵阳-九寨-黄龙-都江堰-成都双飞5日游>","price":"2899"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100011","p_name":"<桂林-阳朔-古东-世外桃源双飞4日游>","price":"2249"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100012","p_name":"<成都-九寨沟-黄龙双飞6日游>","price":"2611"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100013","p_name":"<版纳-香格里拉-昆大丽4飞一卧8日游>","price":"3399"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100014","p_name":"<成都-都江堰-黄龙九寨沟双飞6日游>","price":"2989"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100015","p_name":"<桂林-大漓江-阳朔-龙脊梯田双飞4日游>","price":"2048"}'

put 'data',Date.new().getTime(),'f:content','{"p_id":"100016","p_name":"<长沙-张家界-黄龙洞-天门山-凤凰双飞7日游>","price":"3141"}'

回到Hive终端，创建与HBase对应的Hive表。

create external table data(rowkey string,content string) row format serde 'org.apache.hadoop.hive.hbase.HBaseSerDe' stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' with serdeproperties('hbase.columns.mapping'=':key,f:content') tblproperties ('hbase.table.name'='data');

查看结果

                                select * from data;

创建一个名为datacleans的自定义函数。（如果我们之前退出了hive终端，再次进入的时候记得再重新导一次jar包。否则我们调用函数的时候会报错。）

add jar hdfs://hacluster/user/hive_user/HiveFunction-jar-with-dependencies.jar;
create function datacleans as 'com.huawei.hiveFunctions.UDTFCase';

调用datacleans函数进行拆分

                                select datacleans(content) as (id,name,price) from data;

标签：xxx,hadoop,Hive,MRS,import,apache,org,HBase,hbase
From： https://www.cnblogs.com/playforever/p/17929900.html

MRS基础组件之HBase与Hive开发应用_tyt2023