用户画像 | 标签数据存储之HBase真实应用

create  '<table name>','<column family>'

scan  '<table name>',{LIMIT=>10}

get  '<table name>','row1'

put  '<table name>','row1','<colfamily:colname>','<value>'

put  '<table name>','row ','Column family:column name','new value'

disable  '<table name>'
drop  '<table name>'

# 判断活跃节点
global activenode
for node in ("10.xxx.xx.xxx","10.xxx.xx.xxx"):   # 两台机器作为Master，判断哪台HMaster处于活跃状态
    command = "curl http://"+ str(node) + ":9870/jmx?qry=Hadoop:service=NameNode,name=NameNodeStatus"
    status = os.popen(command).read()
    print("HBase Master status: ".format(status))
    if ("active" in status):
        activenode = node

create 'userprofile_labels', { NAME => "f", BLOCKCACHE => "true" , BLOOMFILTER => "ROWCOL" , COMPRESSION => 'snappy', IN_MEMORY => 'true' }, {NUMREGIONS => 10,SPLITALGO => 'HexStringSplit'}

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.HBase.client.ConnectionFactory
import org.apache.hadoop.HBase.{HBaseConfiguration, KeyValue, TableName}
import org.apache.hadoop.HBase.io.ImmutableBytesWritable
import org.apache.hadoop.HBase.mapreduce.{HFileOutputFormat2, LoadIncrementalHFiles}
import org.apache.hadoop.HBase.util.Bytes
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.sql.SparkSession

object Hive2HBase {
  def main(args: Array[String]): Unit = {
      // 传入日期参数 和 当前活跃的master节点
    val data_date = args(0)
    val node = args(1)   //当前活跃的节点ip

    val spark = SparkSession
      .builder()
      .appName("Hive2HBase")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.storage.memoryFraction", "0.1")
      .config("spark.shuffle.memoryFraction", "0.7")
      .config("spark.memory.useLegacyMode", "true")
      .enableHiveSupport()
      .getOrCreate()
        
       //创建HBase的配置
    val conf = HBaseConfiguration.create()
        conf.set("HBase.zookeeper.quorum", "10.xxx.xxx.xxx,10.xxx.xxx.xxx")
        conf.set("HBase.zookeeper.property.clientPort", "8020")

    //为了预防hfile文件数过多无法进行导入，设置参数值
    conf.setInt("HBase.hregion.max.filesize", 10737418240)
    conf.setInt("HBase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", 3200)
      
    val Data = spark.sql(s"select userid,userlabels from dw.userprofile_usergroup_labels_all where data_date='${data_date}'")
    val dataRdd = Data.rdd.flatMap(row => {    
      val rowkey = row.getAs[String]("userid".toLowerCase)
      val tagsmap = row.getAs[Map[String, Object]]("userlabels".toLowerCase)
      val sbkey = new StringBuffer()  // 对MAP结构转化 a->b  'a':'b'
      val sbvalue = new StringBuffer()
      for ((key, value) <- tagsmap){
        sbkey.append(key + ":")
        val labelght = if (value == ""){
          "-999999"
        } else {
          value
        }
        sbvalue.append(labelght + ":")
      }
      val item = sbkey.substring(0,sbkey.length -1)
      val score = sbvalue.substring(0,sbvalue.length -1)
      Array(
        (rowkey,("f","i",item)),
        (rowkey,("f","s",score))
      )
    })

    // 将rdd转换成HFile需要的格式
    val rdds = dataRdd.filter(x=>x._1 != null).sortBy(x=>(x._1,x._2._1, x._2._2)).map(x => {
      //KeyValue的实例为value
      val rowKey = Bytes.toBytes(x._1)
      val family = Bytes.toBytes(x._2._1)
      val colum = Bytes.toBytes(x._2._2)
      val value = Bytes.toBytes(x._2._3.toString)
      (new ImmutableBytesWritable(rowKey), new KeyValue(rowKey, family, colum, value))
    })

    //文件保存在hdfs的位置
    val locatedir = "hdfs://" + node.toString + ":8020/user/bulkload/hfile/usergroup_HBase_" + data_date

    //在locatedir生成的Hfile文件
    rdds.saveAsNewAPIHadoopFile(locatedir,
      classOf[ImmutableBytesWritable],
      classOf[KeyValue],
      classOf[HFileOutputFormat2],
      conf)
    //HFile导入到HBase
    val load = new LoadIncrementalHFiles(conf)

    //HBase的表名
    val tableName = "userprofile_labels"
    //创建HBase的链接,利用默认的配置文件,读取HBase的master地址
    val conn = ConnectionFactory.createConnection(conf)
    //根据表名获取表
    val table = conn.getTable(TableName.valueOf(tableName))

    try {
      //获取HBase表的region分布
      val regionLocation = conn.getregionLocation(TableName.valueOf(tableName))
      //创建一个hadoop的mapreduce的job
      val job = Job.getInstance(conf)
      //设置job名称，任意命名
      job.setJobName("Hive2HBase")
      //输出文件的内容KeyValue
      job.setMapOutputValueClass(classOf[KeyValue])
      //设置文件输出key, outkey要用ImmutableBytesWritable
      job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
      //配置HFileOutputFormat2的信息
      HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocation)
      //开始导入
      load.doBulkLoad(new Path(locatedir), conn.getAdmin, table, regionLocation)
    } finally {
      table.close()
      conn.close()
    }
    spark.close()
  }
}

# 查询Hive中数据
def check_Hive_data(data_date):
      r = os.popen("Hive -S -e\"select count(1) from dw.userprofile_usergroup_labels_all where data_date='"+data_date+"'\"")
      Hive_userid_count = r.read()
      r.close()
      Hive_count = str(int(Hive_userid_count) 
      print "Hive_result: " + str(Hive_count)
      print "Hive select finished!"

# 查询HBase中数据 
def check_HBase_data(data_date):
      r = os.popen("HBase  org.apache.hadoop.HBase.mapreduce.RowCounter 'userprofile_labels'\" 2>&1 |grep ROWS")
      HBase_count = r.read().strip()[5:]
      r.close()
      print "HBase result: " + str(HBase_count)
      print "HBase select finished!"

# 连接 DB,将查询结果插入表
db = MySQLdb.connect(host="xx.xx.xx.xx",port=3306,user="username", passwd="password", db="xxx", charset="utf8")
cursor = db.cursor()
cursor.execute("INSERT INTO service_monitor(date, service_type, Hive_count, HBase_count) VALUES('"+datestr_+"', 'advertisement', "+str(Hive_userid_count)+","+str(HBase_count)+")")
db.commit()