kafka到hbase API

QQ圖片20200511215405.png

import java.util

import com.ybq.consumer.myutils.{ConnectionInstance, HbaseUtil, PropertiesUtil}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, HTable, Put}
import org.apache.hadoop.hbase.util.Bytes

object HbaseDao {

private val sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
private val sdf2 = new SimpleDateFormat("yyyyMMddHHmmss")
private val cacheList = new util.ArrayList[Put]
//默認讀取resource 下面的文件hbase-site.xml
var conf: Configuration = HBaseConfiguration.create()
private var regions: Integer = Integer.valueOf(PropertiesUtil.getProperty("hbase.calllog.regions"))
private var namespace: String = PropertiesUtil.getProperty("hbase.calllog.namespace")
private var tableName: String = PropertiesUtil.getProperty("hbase.calllog.tablename")
var table: HTable = null
//首先創(chuàng)建命名空間
if(!HbaseUtil.isExisTable(conf,tableName)){
HbaseUtil.initNamespace(conf,namespace)
HbaseUtil.createTable(conf,tableName,regions,"f1","f2")
}
def put(value:String): Unit ={
if(cacheList.size() == 0){
val connection: Connection = ConnectionInstance.getConnection(conf)
table = connection.getTable(TableName.valueOf(tableName)).asInstanceOf[HTable]
table.setAutoFlush(false)
table.setWriteBufferSize(210241024)
}
val splitOri: Array[String] = value.split(",")
val caller: String = splitOri(0)
val callerName: String = splitOri(1)
val callee: String = splitOri(2)
val calleeName: String = splitOri(3)
val buildTime: String = splitOri(4)
val duration: String = splitOri(5)
val flag: String = splitOri(6)
//獲取region編碼
val regionCode: String = HbaseUtil.getRegionCode(caller, buildTime, regions)
//建立通話時長
val buildTimeReplace: String = sdf2.format(sdf1.parse(buildTime))
val buildTimeTs: String = String.valueOf(sdf1.parse(buildTime).getTime)
//生成rowKey
val rowKey: String = HbaseUtil.getRotKey(regionCode, caller, buildTimeReplace, callee, flag, duration)
//向表中插入數(shù)據(jù)
val put = new Put(Bytes.toBytes(rowKey))
//主叫號碼
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call1"), Bytes.toBytes(caller))
//主號名稱
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call1_name"), Bytes.toBytes(callerName))
//被叫號碼
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call2"), Bytes.toBytes(callee))
//被叫名稱
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call2_name"), Bytes.toBytes(calleeName))
//通話日期
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("build_time"), Bytes.toBytes(buildTime))
//通話時間
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("build_time_ts"), Bytes.toBytes(buildTimeTs))
//通過標識
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("flag"), Bytes.toBytes("1"))
//通話時長
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("duration"), Bytes.toBytes(duration))
cacheList.add(put)

if(cacheList.size() >= 0){
  table.put(cacheList)
  table.flushCommits()
  println("插入數(shù)據(jù)成功")
  cacheList.clear()
}

}

object KafkaToHbase {

def main(args: Array[String]): Unit = {

//創(chuàng)建kafka消費者對象
val kafkaConsumer = new KafkaConsumer[String, String](PropertiesUtil.properties)
//訂閱指定的topic 用于數(shù)據(jù)的消費
kafkaConsumer.subscribe(util.Arrays.asList(PropertiesUtil.getProperty("kafka.topics")))
println("等待消費數(shù)據(jù)")
while (true){
  //每0.1s從指定topic中消費數(shù)據(jù)
  val records: ConsumerRecords[String, String] = kafkaConsumer.poll(100)
  //這個scala和java集合類型之間的轉(zhuǎn)換
  for(cr <- records){
    //得到每條數(shù)據(jù)的value
    val str: String = cr.value()
    println(str)
    HbaseDao.put(str)
  }
}

}

object ConnectionInstance {

private var connection: Connection = null

def getConnection(conf:Configuration):Connection={
if(connection == null || connection.isClosed){
connection = ConnectionFactory.createConnection(conf)
}
connection
}

}

object HbaseUtil {
def getRotKey(regionCode: String, caller: String, buildTime: String, callee: String, flag: String, duration: String): String = {
val sb = new StringBuilder
sb.append(regionCode + "")
.append(caller + "")
.append(buildTime + "")
.append(caller + "")
.append(flag + "_")
.append(duration)
sb.toString()
}

def getRegionCode(caller: String, buildTime: String, regions: Integer): String = {
//電話號碼長度
val len: Int = caller.length
//取出后4位號碼
val lastPhone: String = caller.substring(len - 4)
//取出建立通過時間的年月2018-02-02
val ym: String = buildTime.replaceAll("-", "")
.replaceAll(":","")
.replaceAll(" ","")
.substring(0,6)
//離散化操作 1 ^ 這個符號是異或運算對應(yīng)位置相同為0 不同就為1
val x:Integer = Integer.valueOf(lastPhone) ^ Integer.valueOf(ym)
//離散操作
val y: Int = x.hashCode()
//生成分區(qū)號
val regionCode:Int = y % regions
//格式化分區(qū)號
val df = new DecimalFormat("00")

df.format(regionCode)

}

def createTable(conf: Configuration, tableName: String, regions: Integer, columnFamily:String*) ={
val connection: Connection = ConnectionFactory.createConnection(conf)
val admin: Admin = connection.getAdmin
val nameName = new HTableDescriptor(TableName.valueOf(tableName))
for (elem <- columnFamily) {
nameName.addFamily(new HColumnDescriptor(elem))
}
admin.createTable(nameName,getSplitKeys(regions))
}

def getSplitKeys(regions: Integer): Array[Array[Byte]] = {
//定義一個存放區(qū)間的數(shù)組
val keys = new ArrayString
//目前推算region的個數(shù)不會超過兩位數(shù) 格式化區(qū)間
val df = new DecimalFormat("00")
//對region個數(shù)進行遍歷
for(i<-0 until regions){
//使用|拼接一下
keys(i) = df.format(i) + "|"
}
//定義一個二維數(shù)組
val splitKeys = new ArrayArray[Byte]
//比較器升序排序
val treeset = new util.TreeSetArray[Byte]
for(i<-0 until regions){
//使用|拼接一下
treeset.add(Bytes.toBytes(keys(i)))
}

val spiltKeysIterator: util.Iterator[Array[Byte]] = treeset.iterator()
var index = 0
while (spiltKeysIterator.hasNext) {
  val b: Array[Byte] = spiltKeysIterator.next()
  splitKeys(index) = b
  index = index +1
}
splitKeys

}

def initNamespace(conf: Configuration, namespace: String) = {
//獲取hbase連接
val connection: Connection = ConnectionFactory.createConnection(conf)
val admin: Admin = connection.getAdmin
val nd: NamespaceDescriptor = NamespaceDescriptor.create(namespace).addConfiguration("CREATE_TIME", String.valueOf(System.currentTimeMillis()))
.addConfiguration("AUTHOR", "yang").build()
admin.createNamespace(nd)
admin.close()
connection.close()
}

def isExisTable(conf:Configuration,tableName:String): Boolean ={
val connection: Connection = ConnectionFactory.createConnection(conf)
val admin: Admin = connection.getAdmin
val result: Boolean = admin.tableExists(TableName.valueOf(tableName))
admin.close()
connection.close()
result
}

}

object PropertiesUtil {

val is: InputStream = ClassLoader.getSystemResourceAsStream("hbase_consumer.properties")
val properties = new Properties
properties.load(is)

def getProperty(key:String):String={
val str: String = properties.getProperty(key)
str
}

}