工具类-Util
谭彦
HbaseUtil
/**
* @author Alain
* @date 2022/4/14
* @description
*/
object HBaseUtil {
@transient lazy val log = LoggerFactory.getLogger(this.getClass)
private val prop = PropertiesUtil.load("Hbase.properties")
private val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort",prop.getProperty("hbase.zookeeper.property.clientPort") )
conf.set("hbase.zookeeper.quorum", prop.getProperty("hbase.zookeeper.quorum"))
@volatile private var connection: Connection = _
@volatile private var num = 0
//TODO:DML操作
//TODO:获取表连接
def getHBaseConn: Connection = {
synchronized {
if (connection == null || connection.isClosed() || num == 0) {
connection = ConnectionFactory.createConnection(conf)
println("conn is created! " + Thread.currentThread().getName())
}
//每请求一次连接，计数器加一
num = num + 1
println("request conn num: " + num + " " + Thread.currentThread().getName())
}
connection
}
//TODO:关闭表连接
def closeHbaseConn(): Unit = {
synchronized {
if (num <= 0) {
println("no conn to close!")
return
}
//每请求一次关闭连接，计数器减一
num = num - 1
println("request close num: " + num + " " + Thread.currentThread().getName())
//请求连接计数器为0时关闭连接
if (num == 0 && connection != null && !connection.isClosed()) {
connection.close()
println("conn is closed! " + Thread.currentThread().getName())
}
}
}
//TODO:获取表
def getTable(connection: Connection,tableName: String): Table ={
connection.getTable(TableName.valueOf(tableName))
}
//TODO:数据写入Hbase
def putToHBase(conn: Connection, tableName:String, rowKey:String, columnFamily: String, columnIdentifiers: Array[String], values:Array[String]): Unit ={
//获取连接表
val table = classOf[HTable].cast(getTable(conn,tableName))
//封装Hbase对应的put
try {
//设定行键（rowkey）
val put = new Put(Bytes.toBytes(rowKey))
for (i <- 0 to values.length-1){
put.addColumn(columnFamily.getBytes(), columnIdentifiers(i).getBytes(), values(i).getBytes())
}
//执行插入数据到Hbase
table.put(put)
} catch {
case e: Exception => e.printStackTrace()
} finally {
table.close()
}
}
//TODO:rowKey设计 --- 可根据dir取hash，然后拼接time，拼接用户工号为rowKey
def makeRowKey(dir:String,time:String,user:String)={
//将dir进行hash取前10位
val hash: String = dir.hashCode.toString
var endLength=10
if (hash.length < 10) {
endLength=hash.length
}
//拼接rowKey：
val rowKey= dir.hashCode.toString.substring(0,endLength)+time+user
rowKey
}
//TODO:通过rowKey进行scan数据
def scanAndParseLongFromHBase(conn: Connection, tableName: String, columnFamily: String, columnIdentifier: Array[String], startRow: String,endRow :String): ListBuffer[JSONObject] ={
val scanResult: ListBuffer[JSONObject] = new ListBuffer[JSONObject]
//1、获取表连接
val table = classOf[HTable].cast(getTable(conn,tableName))
//2、创建scan对象 + 进行扫描的过滤
val scan: Scan = new Scan()
scan.withStartRow(startRow.getBytes())
scan.withStopRow(endRow.getBytes())
for (elem <- columnIdentifier) {
scan.addColumn(columnFamily.getBytes(),elem.getBytes())
}
try {
//3、扫描数据
val results: ResultScanner = table.getScanner(scan)
//4、解析result
val it: util.Iterator[Result] = results.iterator()
while (it.hasNext){
val result: Result = it.next()
val jSONObject: JSONObject = new JSONObject()
for (elem <- columnIdentifier) {
jSONObject.put(elem,Bytes.toString(result.getValue(columnFamily.getBytes(),elem.getBytes())))
}
//结果加到返回结果集中
scanResult.append(jSONObject)
}
scanResult
} catch {
case e: Exception => e.printStackTrace()
scanResult
} finally {
table.close()
}
}
def getFromHBase(connection: Connection, tableName: String, columnFamily: String, columnIdentifier: String, rowKey: String): String ={
var value = ""
val table = connection.getTable(TableName.valueOf(tableName))
val get = new Get(rowKey.getBytes())
get.addColumn(columnFamily.getBytes(),columnIdentifier.getBytes())
val result = table.get(get)
if(table != null) table.close()
if (!result.isEmpty) {
value = Bytes.toString(result.getValue(columnFamily.getBytes(),columnIdentifier.getBytes()))
}
value
}
def getAndParseLongFromHBase(connection: Connection, tableName: String, columnFamily: String, columnIdentifier: String, rowKey: String): Long ={
var value = ""
var reslut = 0L
val table = connection.getTable(TableName.valueOf(tableName))
val get = new Get(rowKey.getBytes())
get.addColumn(columnFamily.getBytes(),columnIdentifier.getBytes())
val result = table.get(get)
if(table != null) table.close()
if (!result.isEmpty) {
value = Bytes.toString(result.getValue(columnFamily.getBytes(),columnIdentifier.getBytes()))
}
try {
reslut = java.lang.Long.valueOf(value)
} catch {
case exception: Exception => log.info("No data for rowKey:{} , table name:{} , column family:{} , identifier:{}",rowKey,tableName,columnFamily,columnIdentifier)
//log.info("No data for rowKey: " + rowKey + "table name: " + tableName + "column family: " + columnFamily + "identifier：" + columnIdentifier)
}
reslut
}
def getFromHBaseTable(conn: Connection,tableName:String,rowKey:String,columnFamily: String, columnIdentifier: String,value:Array[Byte]): String ={
val table = classOf[HTable].cast(conn.getTable(TableName.valueOf(tableName)))
val get =new Get(rowKey.getBytes())
val value = Bytes.toString(table.get(get).getValue(columnFamily.getBytes(),columnIdentifier.getBytes()))
if(table!=null){
table.close()
}
value
}
//TODO:DDL操作
//TODO 判断表是否存在
def isTableExist(tableName: String,conn:Connection): Boolean = {
//1.获取DDL操作对象
val admin: Admin = conn.getAdmin
try {
//2.判断表是否存在操作
val exists: Boolean = admin.tableExists(TableName.valueOf(tableName))
//3.返回结果
exists
} catch {
case e: Exception => e.printStackTrace()
false
} finally {
//4.关闭资源
admin.close()
}
}
//TODO:直连方式创建Hbase表
def createTable(tableName: String, cfs: String*): Unit = {
//1.判断是否存在列族信息
if (cfs.length <= 0) {
System.out.println("请设置列族信息！")
return
}
//2.获取与HBase的连接
val conn: Connection = getHBaseConn
//3.判断表是否存在（如果查询失败表示不存在该表）
if (isTableExist(tableName,conn)) {
System.out.println("需要创建的表已存在！")
return
}
//4.获取DDL操作对象
val admin: Admin = conn.getAdmin
try {
//5.创建表描述器构造器
val tableDescriptorBuilder: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
//6.循环添加列族信息
for (cf <- cfs) {
val columnFamilyDescriptorBuilder: ColumnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf))
tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build)
}
//7.执行创建表的操作
admin.createTable(tableDescriptorBuilder.build)
} catch {
case e: Exception => e.printStackTrace()
} finally {
//8.关闭资源
admin.close()
conn.close()
}
}
}
文件读取propertiesUtil
/**
* @author Alain
* @date 2022/4/14
* @description 加载类路径的一个properties文件，只需要调用load方法，
* 自动将properties文件读取，并封装一个Properties返回
*/
object PropertiesUtil {
private val properties = new Properties
{
properties.
load(new InputStreamReader(Thread.currentThread().getContextClassLoader.getResourceAsStream("system.properties") , "UTF-8"))
}
def load(propertieName:String): Properties ={
properties.load(new InputStreamReader(Thread.currentThread().getContextClassLoader.getResourceAsStream(propertieName) , "UTF-8"))
properties
}
//测试
def main(args: Array[String]): Unit = {
// val properties: Properties = load("config.properties")
val prop: Properties = PropertiesUtil.load("kafka.properties")
println(prop)
println(prop.get("sasl.jaas.config"))
println(prop.get("hs2.connection.url"))
}
}
java 版本
/**
* Created by Alains on 2021/9/3 15:07
*
* 加载类路径的一个properties文件，只需要调用load方法，自动将properties文件读取，
* 封装一个Properties返回
*/
public class PropertiesUtil {
private static Properties properties = new Properties();
static {
try {
ClassLoader classLoader = PropertyTools.class.getClassLoader();
InputStream is = classLoader.getResourceAsStream("system.properties");
properties.load(is);
} catch (IOException e) {
e.printStackTrace();
}
}
/* public static void load(String propName) throws IOException {
properties.load(new InputStreamReader(Thread.currentThread().getContextClassLoader().getResourceAsStream(propName) , "UTF-8"));
}*/
public static void load(String propName){
try {
ClassLoader classLoader = PropertyTools.class.getClassLoader();
InputStream is = classLoader.getResourceAsStream(propName);
properties.load(is);
} catch (IOException e) {
e.printStackTrace();
}
}
}MD5-Util 获取加密数据--scala版本
/**
* @author Alain
* @date 2022/4/15
* @description
*/
object MD5 {
def hashMD5(content: String): String = {
val md5 = MessageDigest.getInstance("MD5")
val encoded = md5.digest((content).getBytes)
encoded.map("%02x".format(_)).mkString
}
def main(args: Array[String]): Unit = {
val param = ("ALL","ALL","ALL","ALL","ALL","ALL","ALL").productIterator.mkString(",")
println(param)
val md5 = hashMD5(param)
println("20220101_1_"+md5) //20220101_1_b9af90d9b37ac0de6c0098d2dc118a75
}
}MD5-Util 获取加密数据--java版本
public class Utils {
/**
* MD5加密
* @param content
* @return
*/
public static String hashMD5( String content) {
try {
// 指定MD5加密算法
MessageDigest md5 = MessageDigest.getInstance("MD5");
// 将字符串中转换成byte数组,并更新到MD5
md5.update(content.getBytes());
// 对输入数据进行随机哈希加密
byte[] encoded = md5.digest();
// 转换为十六进制字符串
return getHex(encoded);
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
}
return null;
}
/**
* 转换为十六进制字符串
* 将加密后的每个字节转化成十六进制，一个字节8位，相当于2个16进制，不足2位的前面补0
*/
private static String getHex(byte[] bytes){
StringBuilder builder = new StringBuilder(bytes.length * 2);
for (byte b: bytes) {
builder.append(Integer.toHexString((b >> 4) & 0x0f));
builder.append(Integer.toHexString(b & 0x0f ));
}
return builder.toString();
}
}bean类型封装到 --》 另一个bean类型
/**
* Created by Alains on 2022/4/17 17:20
*/
import java.lang.reflect.{Field, Modifier}
import java.text.SimpleDateFormat
import java.util.Locale
import scala.util.control.Breaks._
object MyBeanUtils {
/**
* 对象相同属性copy
*
* @param obj
* @param toResult
* @return
* @throws Exception
* 转换报错
*/
def copyProperties(obj: Object, toResult: Object): Unit = {
if (obj == null) {
return
}
try {
val fields = toResult.getClass.getDeclaredFields
for (field <- fields) {
breakable {
field.setAccessible(true) //修改访问权限
if (Modifier.isFinal(field.getModifiers()))
break
if (isWrapType(field)) {
val getMethodName = field.getName()
val setMethodName = field.getName()+"_$eq"
val getMethod = {
try {
obj.getClass().getMethod(getMethodName)
} catch {
case ex: Exception =>
// println(ex)
break
}
}
//从源对象获取get方法
val setMethod =
toResult.getClass.getMethod(setMethodName, field.getType)
//从目标对象获取set方法
val value = {
val objValue = getMethod.invoke(obj) // get 获取的是源对象的值
if (objValue != null && objValue.isInstanceOf[java.util.Date]) {
// GMT时间转时间戳
val format =
new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.ENGLISH)
val formatValue = format.format(objValue)
java.lang.Long.valueOf(format.parse(formatValue).getTime)
} else
objValue
}
setMethod.invoke(toResult, value)
}
}
}
} catch {
case ex: Exception => throw ex
}
}
/**
* 是否是基本类型、包装类型、String类型
*/
def isWrapType(field: Field): Boolean = {
val typeList = List[String]("java.lang.Integer"
, "java.lang.Double", "java.lang.Float", "java.lang.Long", "java.util.Optional"
, "java.lang.Short", "java.lang.Byte", "java.lang.Boolean", "java.lang.Char"
, "java.lang.String", "int", "double", "long"
, "short", "byte", "boolean", "char", "float")
if (typeList.contains(field.getType().getName())) true else false
}
def getV(ref:AnyRef, name: String): Any = ref.getClass.getMethods.find(_.getName == name).get.invoke(ref)
def setV(ref:AnyRef,name: String, value: Any): Unit = {
ref.getClass.getMethods.find(_.getName == name + "_$eq").get.invoke(ref, value.asInstanceOf[AnyRef])
}
}RedisUtil
/**
* 〈一句话功能简述〉
* 〈获取维度数据，并进行过滤〉
*/
@Slf4j
public class RedisUtil {
private static final long serialVersionUID = 1L;
/**配置信息*/
private static final String REDIS_NONE_VALUE_DEFAULT = "";
private static Cache<String, String> REDIS_VALUE_CACHE = CacheBuilder.newBuilder()
.concurrencyLevel(32).initialCapacity(5000000)
.expireAfterWrite(60, TimeUnit.MINUTES).build();
/**接口项*/
private static final Map<String, JedisCluster> REDIS_TABLE_MAP = new ConcurrentSkipListMap<>();
public RedisUtil() {
}
/** redis client 初始化 */
public static JedisCluster init(Set<HostAndPort> redisConf) throws IOException {
String redisClientKey = redisConf.toString();
//初始化hbase
JedisCluster jedisCluster = null;
if(REDIS_TABLE_MAP.containsKey(redisClientKey)){
jedisCluster = REDIS_TABLE_MAP.get(redisClientKey);
}else {
jedisCluster = new JedisCluster(redisConf);
REDIS_TABLE_MAP.put(redisClientKey, jedisCluster);
}
return jedisCluster;
}
/**
* redis get 数据 通过 rowKey
* */
public static String get(Set<HostAndPort> redisConf, String key) throws IOException {
String redisClientKey = redisConf.toString();
//初始化hbase
JedisCluster jedisCluster = init(redisConf);
//redis查询
try {
String value = jedisCluster.get(key);
LogPrintKVControl.sample(key, String.format("redis data: %s", value),100);
return value;
} catch (Exception e) {
REDIS_TABLE_MAP.remove(redisClientKey);
log.error(String.format("sendDataInredis -- value: %s, error: %s", key, e.getMessage()));
}
return REDIS_NONE_VALUE_DEFAULT;
}
/**
* redis get 数据 通过 rowKey 缓存方式
* */
public static String get(Set<HostAndPort> redisConf, String key, Boolean CacheEnable) throws IOException {
String valueCachekey = key;
String cacheValue = REDIS_VALUE_CACHE.getIfPresent(valueCachekey);
//是否使用缓存数据
if( CacheEnable && cacheValue != null){
//缓存命中
return cacheValue;
}
//缓存未命中
String redsivalue = get(redisConf, key);
//缓存数据更新
REDIS_VALUE_CACHE.put(valueCachekey, redsivalue);
return redsivalue;
}
/**
* redis hget 数据 通过 rowKey
* */
public static String hget(Set<HostAndPort> redisConf, String key, String field) throws IOException {
String redisClientKey = redisConf.toString();
//初始化hbase
JedisCluster jedisCluster = init(redisConf);
//redis查询
try {
String value = Optional.ofNullable(jedisCluster.hget(key, field)).orElse(REDIS_NONE_VALUE_DEFAULT);
REDIS_VALUE_CACHE.put(key + field, value);
LogPrintKVControl.sample(key + field, String.format("redis data: %s", value),100);
return value;
} catch (Exception e) {
REDIS_TABLE_MAP.remove(redisClientKey);
log.error(String.format("send Data In redis -- value: %s, error: %s", key, e.getMessage()));
}
return REDIS_NONE_VALUE_DEFAULT;
}
/**
* redis hget 数据 通过 rowKey 缓存方式
* */
public static String hget(Set<HostAndPort> redisConf, String key, String field, Boolean CacheEnable) throws IOException {
String valueCacheKey = key + field;
String cacheValue = REDIS_VALUE_CACHE.getIfPresent(valueCacheKey);
//是否使用缓存数据
if( CacheEnable && cacheValue != null){
//缓存命中
return cacheValue;
}
//缓存未命中
String redisValue = hget(redisConf, key, field);
return redisValue;
}
/**
* redis hset数据
* */
public static Long hset(Set<HostAndPort> redisConf, String key, String field, String value) throws IOException {
//初始化hbase
JedisCluster jedisCluster = init(redisConf);
Long hset = jedisCluster.hset(key, field, value);
return hset;
}
/**
* redis client 销毁
* */
public static void close(Set<HostAndPort> redisConf) throws IOException {
String redisClientKey = redisConf.toString();
//初始化hbase
if(REDIS_TABLE_MAP.containsKey(redisClientKey)){
JedisCluster jedisCluster = REDIS_TABLE_MAP.get(redisClientKey);
jedisCluster.close();
REDIS_TABLE_MAP.remove(redisClientKey);
}
}
}
类似资料：