Spark：DataFrame寫HFile （Hbase）一個(gè)列族、一個(gè)列擴(kuò)展一個(gè)列族、多個(gè)列

擴(kuò)展:Spark:DataFrame生成HFile 批量導(dǎo)入Hbase

在上一篇博文中遺留了一個(gè)問題，就是只能處理DataFrame 的一行一列,雖然給出一個(gè)折中的辦法處理多個(gè)列，但是對于字段多的DataFrame卻略顯臃腫，經(jīng)過我的研究，實(shí)現(xiàn)了從一個(gè)列族、一個(gè)列到一個(gè)列族、多個(gè)列擴(kuò)展。
此文章再此記錄實(shí)現(xiàn)方法

實(shí)現(xiàn)思路：

保存為HFile的關(guān)鍵是下面這個(gè)方法

saveAsNewAPIHadoopFile(save_path,
      classOf[ImmutableBytesWritable],
      classOf[KeyValue],
      classOf[HFileOutputFormat2],
      job.getConfiguration)

要使用這個(gè)方法就要保證最后的結(jié)果數(shù)據(jù)需要是RDD[(ImmutableBytesWritable, KeyValue)]類型的，所以這就是我們努力前進(jìn)的方向。在這個(gè)過程中有幾個(gè)問題需要解決
1. 如何一次處理DataFrame 的眾多字段

    val result1: RDD[(ImmutableBytesWritable, Seq[KeyValue])] = resultDataFrame
      .map(row => {
        var kvlist: Seq[KeyValue] = List()
        var rowkey: Array[Byte] = null
        var cn: Array[Byte] = null
        var v: Array[Byte] = null
        var kv: KeyValue = null
        val cf: Array[Byte] = clounmFamily.getBytes //列族
        rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
        for (i <- 1 to (columnsName.length - 1)) {
          cn = columnsName(i).getBytes() //列的名稱
          v = Bytes.toBytes(row.getAs[String](columnsName(i))) //列的值
          //將rdd轉(zhuǎn)換成HFile需要的格式,我們上面定義了Hfile的key是ImmutableBytesWritable,那么我們定義的RDD也是要以ImmutableBytesWritable的實(shí)例為key
          kv = new KeyValue(rowkey, cf, cn, v) //封裝一下 rowkey, cf, clounmVale, value
          //
          kvlist = kvlist :+ kv //將新的kv加在kvlist后面（不能反 需要整體有序）
        }
        (new ImmutableBytesWritable(rowkey), kvlist)
      })
  1. 上述代碼中通過map取出每一行row，用一個(gè)for循環(huán)通過所有字段的名稱（去除掉“key”這個(gè)字段）對每個(gè)字段進(jìn)行封裝處理，每處理完一個(gè)字段加入kvlist。
  2. 在此處有個(gè)地方需要注意的是，我們要保證 kvlist 里面的數(shù)據(jù)整體有序（升序），這里的有序由字段名稱排序和加入 kvlist 的位置來保證，kvlist 通過 :+ 將后一個(gè)數(shù)據(jù)放在List的后面，至于字段名稱排序在后面說明。
  3. 至于此處為什么要去除掉key，這是因?yàn)槲夷J(rèn)DataFrame第一個(gè)字段就是key，因?yàn)樾枰獙λ凶侄蚊Q進(jìn)行排序，如果不把key拿出來后續(xù)不知道key在哪里了，如果按照正常走下去，key值也會被當(dāng)成value被保存一次，這顯然不符合我們的要求，當(dāng)然有興趣的同學(xué)可以自己實(shí)現(xiàn)更全面的方法。

2. 如何對DataFrame 的所有字段名排序

 var columnsName: Array[String] = resultDataFrame.columns //獲取列名 第一個(gè)為key
    columnsName = columnsName.drop(1).sorted //把key去掉  因?yàn)橐判?
  1. 通過resultDataFrame.columns獲取所有列名，通過drop(1)刪掉“key”,(序號從1開始)
  2. 通過sorted 對列名進(jìn)行排序，默認(rèn)就是升序的

通過上面方法處理后數(shù)據(jù)類型是
RDD[(ImmutableBytesWritable, Seq[KeyValue])]
這顯然不是我們需要的，但是距離
RDD[(ImmutableBytesWritable, KeyValue)]
已然不遠(yuǎn)矣

3. 如何將value的Seq[KeyValue] 穿換成 KeyValue

 val result: RDD[(ImmutableBytesWritable, KeyValue)] = result1.flatMapValues(s => {
      s.iterator
    })
  1. 這點(diǎn)其實(shí)很簡單，但是腦子當(dāng)時(shí)短路還糾結(jié)很久，直接用flatMapValues這個(gè)方法即可，最后處理出來的就是我們的目標(biāo)RDD[(ImmutableBytesWritable, KeyValue)]

4. 目標(biāo)路徑已經(jīng)存在怎么辦

  /**
    * 刪除hdfs下的文件
    *
    * @param url 需要?jiǎng)h除的路徑
    */
  def delete_hdfspath(url: String) {
    val hdfs: FileSystem = FileSystem.get(new Configuration)
    val path: Path = new Path(url)
    if (hdfs.exists(path)) {
      val filePermission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.READ)
      hdfs.delete(path, true)
    }
  }
  1. 存在就刪除唄，新建個(gè)方法delete_hdfspath將路徑刪除即可

5. 如何生成 HFile 和 load 數(shù)據(jù)到Hbase
執(zhí)行方法saveAsNewAPIHadoopFile()生成HFile
注意：此處要對key進(jìn)行排序（升序）

    //保存數(shù)據(jù)
    result
      .sortBy(x => x._1, true) //要保持 整體有序
      .saveAsNewAPIHadoopFile(save_path,
      classOf[ImmutableBytesWritable],
      classOf[KeyValue],
      classOf[HFileOutputFormat2],
      job.getConfiguration)

load 數(shù)據(jù)到Hbase

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles hdfs://ns1/user/hive/warehouse/iptv.db/zzzHFile iptv:spark_test

過程中出現(xiàn)的問題

  1. DataFrame 字段名稱沒有排序處理
18/10/15 14:19:32 WARN scheduler.TaskSetManager: Lost task 0.1 in stage 2.0 (TID 3, iptve2e03): java.io.IOException: Added a key not lexically larger than previous. 
Current cell = 200_\xE5\xB9\xBF\xE5\xB7\x9E_GD_GZ/cf_info:area_code/1539584366048/Put/vlen=5/seqid=0, 
    lastCell = 200_\xE5\xB9\xBF\xE5\xB7\x9E_GD_GZ/cf_info:dict_id/1539584366048/Put/vlen=2/seqid=0

上面的意思是當(dāng)前列名cf_info:area_code比前一個(gè)列名cf_info:dict_id小，這就是為什么需要對列名排序的原因，同時(shí)還要把key刪除掉，因?yàn)椴粍h除會出現(xiàn)cf_info:key這個(gè)列

完整代碼

依賴：sbt

libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.spark" % "spark-sql_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.spark" % "spark-hive_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.2.0-cdh5.7.2"
libraryDependencies += "org.apache.hbase" % "hbase-protocol" % "1.2.0-cdh5.7.2"
libraryDependencies += "mysql" % "mysql-connector-java" % "5.1.38"
libraryDependencies += "org.apache.spark" % "spark-streaming_2.10" % "1.6.0-cdh5.7.2"
libraryDependencies += "com.yammer.metrics" % "metrics-core" % "2.2.0"

關(guān)鍵代碼

import java.text.SimpleDateFormat
import java.util.{Calendar, Date}

import com.iptv.domain.DatePattern
import com.iptv.job.JobBase
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions.{concat, lit}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
  /**
    * 將DataFrame 保存為 HFile
    *
    * @param resultDataFrame 需要保存為HFile的 DataFrame，DataFrame的第一個(gè)字段必須為"key"
    * @param clounmFamily 列族名稱（必須在Hbase中存在，否則在load數(shù)據(jù)的時(shí)候會失?。?    * @param save_path HFile的保存路徑
    */
  def saveASHfFile(resultDataFrame: DataFrame, clounmFamily: String, save_path: String): Unit = {
    val conf: Configuration = HBaseConfiguration.create()
    lazy val job = Job.getInstance(conf)
    job.setMapOutputKeyClass(classOf[ImmutableBytesWritable]) //設(shè)置MapOutput Key Value 的數(shù)據(jù)類型
    job.setMapOutputValueClass(classOf[KeyValue])

    var columnsName: Array[String] = resultDataFrame.columns //獲取列名 第一個(gè)為key
    columnsName = columnsName.drop(1).sorted //把key去掉  因?yàn)橐判?
    val result1: RDD[(ImmutableBytesWritable, Seq[KeyValue])] = resultDataFrame
      .map(row => {
        var kvlist: Seq[KeyValue] = List()
        var rowkey: Array[Byte] = null
        var cn: Array[Byte] = null
        var v: Array[Byte] = null
        var kv: KeyValue = null
        val cf: Array[Byte] = clounmFamily.getBytes //列族
        rowkey = Bytes.toBytes(row.getAs[String]("key")) //key
        for (i <- 1 to (columnsName.length - 1)) {
          cn = columnsName(i).getBytes() //列的名稱
          v = Bytes.toBytes(row.getAs[String](columnsName(i))) //列的值
          //將rdd轉(zhuǎn)換成HFile需要的格式,我們上面定義了Hfile的key是ImmutableBytesWritable,那么我們定義的RDD也是要以ImmutableBytesWritable的實(shí)例為key
          kv = new KeyValue(rowkey, cf, cn, v) //封裝一下 rowkey, cf, clounmVale, value
          //
          kvlist = kvlist :+ kv //將新的kv加在kvlist后面（不能反 需要整體有序）
        }
        (new ImmutableBytesWritable(rowkey), kvlist)
      })

    delete_hdfspath(save_path) //刪除save_path 原來的數(shù)據(jù)

    //RDD[(ImmutableBytesWritable, Seq[KeyValue])] 轉(zhuǎn)換成 RDD[(ImmutableBytesWritable, KeyValue)]
    val result: RDD[(ImmutableBytesWritable, KeyValue)] = result1.flatMapValues(s => {
      s.iterator
    })

    //保存數(shù)據(jù)
    result
      .sortBy(x => x._1, true) //要保持 整體有序
      .saveAsNewAPIHadoopFile(save_path,
      classOf[ImmutableBytesWritable],
      classOf[KeyValue],
      classOf[HFileOutputFormat2],
      job.getConfiguration)

  }

  /**
    * 刪除hdfs下的文件
    * @param url 需要?jiǎng)h除的路徑
    */
  def delete_hdfspath(url: String) {
    val hdfs: FileSystem = FileSystem.get(new Configuration)
    val path: Path = new Path(url)
    if (hdfs.exists(path)) {
      val filePermission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.READ)
      hdfs.delete(path, true)
    }
  }

使用示例

package com.iptv.job.basedata

import com.iptv.job.JobBase
import org.apache.spark.sql.functions.{concat, lit}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author 利伊奧克兒-lillcol
  *         2018/10/14-11:08
  *
  */
object TestHFile extends JobBase {
  var hdfsPath: String = ""
  var proPath: String = ""
  var DATE: String = ""

  val sparkConf: SparkConf = new SparkConf().setAppName(getClass.getSimpleName)
  val sc: SparkContext = new SparkContext(sparkConf)
  val sqlContext: SQLContext = getSQLContext(sc)

  import sqlContext.implicits._

  def main(args: Array[String]): Unit = {
    hdfsPath = args(0)
    proPath = args(1)

    //HFile保存路徑
    val save_path: String = hdfsPath + "zzzHFile"
    //獲取測試DataFrame
    val dim_sys_city_dict: DataFrame = readMysqlTable(sqlContext, "DIM_SYS_CITY_DICT", proPath)

    val resultDataFrame: DataFrame = dim_sys_city_dict
      .select(concat($"city_id", lit("_"), $"city_name", lit("_"), $"city_code").as("key"), $"*")
    //注：resultDataFrame 里面的 key 要放在第一位，因?yàn)楹竺嫘枰獙ψ侄蚊判?    saveASHfFile(resultDataFrame, "cf_info", save_path)
  }
}

上述讀取mysql數(shù)據(jù)為ＤataFrame的放大可以參考
Spark:讀取mysql數(shù)據(jù)作為DataFrame

此為個(gè)人工作過程中的總結(jié)，轉(zhuǎn)載請標(biāo)出處?。。。?！

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容