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

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

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

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

创建自定义的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函数

                                cd ~/eclipse-workspace/HiveFunction/target/

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

scp ~/eclipse-workspace/HiveFunction/target/HiveFunction-jar-with-dependencies.jar root@xxx.xxx.xxx.xxx:/root

ssh root@xxx.xxx.xxx.xxx

                                vim students

                                beeline
                            

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

select * from students;

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

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

select lower_upper(name) from students;

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

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

                                select mysum(1) from students;

                                ssh root@xxx.xxx.xxx.xxx
                            

                                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

./install.sh /opt/client

cd /opt/client

source bigdata_env

                                hbase shell

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

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;

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

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