lillcol

背景

之前的博客:Spark:DataFrame写HFile (Hbase)一个列族、一个列扩展一个列族、多个列

用spark 1.6.0 和 hbase 1.2.0 版本实现过spark BulkLoad Hbase的功能,并且扩展了其只能操作单列的不便性。

现在要用spark 2.3.2 和 hbase 2.0.2 来实现相应的功能;
本以为会很简单,两个框架经过大版本的升级,API变化很大;
官网的案例其实有点难实现,且网上的资料要么老旧,要么复制黏贴实在是感人,所以花了点时间重新实现了该功能;
同时记录了在这个过程中遇到的很多问题。


版本信息

工具 版本
spark 2.3.2
hbase 2.0.2

配置文件

hdfs.properties
# zookeeper的信息
zk=slave01:2181,slave02:2181,slave03:2181,slave04:2181,slave05:2181
zk.host=slave01,slave02,slave03,slave04,slave05
zk.port=2181

maven 依赖

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <scala.version>2.11</scala.version>
        <spark.version>2.3.2</spark.version>
        <hbase.version>2.0.2</hbase.version>
        <hadoop.version>3.1.1</hadoop.version>
    </properties>
    
    <dependencies>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-it</artifactId>
            <version>${hbase.version}</version>
        </dependency>

    </dependencies>

实现代码

模版方法
package com.aaa.base

import org.apache.spark.storage.StorageLevel
import org.slf4j.{Logger, LoggerFactory}

/**
 * @author lillcol 
 *         create_time  2019/6/14-14:25
 *         description :使用模板方法模式创建任务执行流程,保证所有任务的流程统一,所有非流处理任务需要实现此接口
 */
trait ExportToHbaseTemplate {
  val logger: Logger = LoggerFactory.getLogger(getClass.getSimpleName)
  //任务状态
  val PERSIST_LEVEL: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER


  /**
   * 任务模板
   *
   * @param args
   */
  def runWork(args: Array[String]): Unit = {
    try {
      //      initTepmlate(args) // 模板初始化信息
      init(args) // 初始化信息
      //      printfTepmlate //输出模板初始化结果
      printf //输出初始化结果
      workFlow //数据处理流
    } catch {
      case e: Exception =>
        e.printStackTrace
    } finally {
      //      spark.sparkContext.stop()
    }
  }


  /**
   * 初始化信息
   *
   * @param args
   */
  def init(args: Array[String])

  /**
   * 输出初始化结果
   */
  def printf()

  /**
   * 数据处理流
   */
  def workFlow()

  /**
   * 模板初始化
   *
   * @param args
   */
  def initTepmlate(args: Array[String]): Unit = {
  }

  /**
   * 输出模板初始化结果
   */
  def printfTepmlate(): Unit = {
  }

}


读取配置文件方法
package com.aaa.util

import java.io.FileInputStream
import java.util.Properties

/**
 * 读取.properties配置文件
 *
 * @param path
 */
class ReadProperties(path: String) {
  /**
   * 读取、加载指定路径配置文件
   *
   * @return Properties 实例
   */
  def getProInstance(): Properties = {
    val pro = new Properties()
    pro.load(new FileInputStream(path))
    pro
  }
}

/**
 * 伴生对象
 */
object ReadProperties {
  def getInstance(path: String): ReadProperties = {
    new ReadProperties(path)
  }
}


实现主体
package com.aaa.test

import com.aaa.base.{ExportToHbaseTemplate}
import com.aaa.util.ReadProperties
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, RegionLocator, Table}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2
import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue, TableName}
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

object TestHbase extends ExportToHbaseTemplate {

  val proPath = "/root/lillcol/hdfs.properties" //配置文件路径
  var cf: String = "info" //列族
  var defKey: String = "UID" //默认key
  val proInstance = ReadProperties.getInstance(proPath).getProInstance
  var partition: String = "20190918"
  var conf: Configuration = _
  var SourceDataFrame: DataFrame = _
  var outPutTable: String = "outPutTable"
  var savePath: String = s"/tmp/hbase/$outPutTable" //临时HFile保存路径
  val spark: SparkSession = SparkSession
    .builder()
    //    .master("local")
    .appName("ExportToHBase")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .getOrCreate()

  import spark.implicits._

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

  /**
   * 初始化信息
   *
   * @param args
   */
  override def init(args: Array[String]): Unit = {
    conf = HBaseConfiguration.create() //Hbase配置信息
    conf.set("hbase.zookeeper.quorum", proInstance.getProperty("zk")) //Hbase zk信息
    conf.set("hbase.mapreduce.hfileoutputformat.table.name", outPutTable) //Hbase 输出表
    conf.set("hbase.unsafe.stream.capability.enforce", "false") //hbase  根目录设定  (有时候会报错,具体看错误处理部分)
    conf.set("zookeeper.znode.parent", "/hbase")
    conf.set("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", "400")
  }

  /**
   * 数据处理流
   */
  override def workFlow(): Unit = {
    getDataset()
    val hfileRDD: RDD[(ImmutableBytesWritable, KeyValue)] = getHFileRDD
    saveHfile(hfileRDD)
    loadHFileToHbase()
  }

  /**
   * 获取源数据表
   */
  def getDataset() = {
    SourceDataFrame = spark.read.parquet("/warehouse/data/lillcol/test.parquet")
  }

  /**
   * 将dataset处理成Hbase的数据格式
   * 注:
   * 默认API只能处理一个列族一个列的情况
   * 此处扩展了该功能:
   * 用var kvlist: Seq[KeyValue] = List()
   * 和rdd.flatMapValues(_.iterator) 方式自适应列名
   * 处理后的结果为:一个列族多个列
   *
   * @return
   */
  def getHFileRDD(): RDD[(ImmutableBytesWritable, KeyValue)] = {
    //key:全局变量不能在 map  内部使用  所以创建一个局部变量
    //注:如果不做会出现奇怪的异常比如类初始化失败,spark为初始化等,目前还没发现具体原因,后续去跟踪
    val key = defKey
    //列族
    val clounmFamily: String = cf
    //获取列名 第一个为key
    val columnsName: Array[String] = SourceDataFrame.columns.sorted

    val result1: RDD[(ImmutableBytesWritable, Seq[KeyValue])] = SourceDataFrame
      .repartition(200, $"$key") //如果数据量大,可以根据key进行分区操作
      .rdd
      .map(row => {
        var kvlist: Seq[KeyValue] = List() //存储多个列
        var kv: KeyValue = null
        val cf: Array[Byte] = clounmFamily.getBytes //列族
        val rowKey = Bytes.toBytes(row.getAs[Int](key) + "")
        val immutableRowKey: ImmutableBytesWritable = new ImmutableBytesWritable(rowKey)
        for (i <- 0 to (columnsName.length - 1)) {
          //将rdd转换成HFile需要的格式,
          //我们上面定义了Hfile的key是ImmutableBytesWritable,
          //那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
          var value: Array[Byte] = null
          try {
            //数据是字符串的都映射成String
            value = Bytes.toBytes(row.getAs[String](columnsName(i)))
          } catch {
            case e: ClassCastException =>
              //出现数据类型转换异常则说明是数字,都映射成BigInt
              value = Bytes.toBytes(row.getAs[BigInt](columnsName(i)) + "")
            case e: Exception =>
              e.printStackTrace()
          }
          //封装KeyValue
          kv = new KeyValue(rowKey, cf, Bytes.toBytes(columnsName(i)), value)
          //将新的kv加在kvlist后面(不能反 需要整体有序)
          kvlist = kvlist :+ kv
        }
        (immutableRowKey, kvlist)
      })

    val hfileRDD: RDD[(ImmutableBytesWritable, KeyValue)] = result1
      .flatMapValues(_.iterator)
    hfileRDD
  }

  /**
   * 保存生成的HFile文件
   * 注:bulk load  生成的HFile文件需要落地
   * 然后再通过LoadIncrementalHFiles类load进Hbase
   * 此处关于  sortBy 操作详解:
   * 0. Hbase查询是根据rowkey进行查询的,并且rowkey是有序,
   * 某种程度上来说rowkey就是一个索引,这是Hbase查询高效的一个原因,
   * 这就要求我们在插入数据的时候,要插在rowkey该在的位置。
   * 1. Put方式插入数据,会有WAL,同时在插入Hbase的时候会根据RowKey的值选择合适的位置,此方式本身就可以保证RowKey有序
   * 2. bulk load 方式没有WAL,它更像是hive通过load方式直接将底层文件HFile移动到制定的Hbase路径下,所以,在不东HFile的情况下,要保证本身有序才行
   * 之前写的时候只要rowkey有序即可,但是2.0.2版本的时候发现clounm也要有序,所以会有sortBy(x => (x._1, x._2.getKeyString), true)
   *
   * @param hfileRDD
   */
  def saveHfile(hfileRDD: RDD[(ImmutableBytesWritable, KeyValue)]) = {
    //删除可能存在的文件,否则回报文件已存在异常
    delete_hdfspath(savePath)

    //生成的HFile保存到指定目录
    hfileRDD
      .sortBy(x => (x._1, x._2.getKeyString), true) //要保持 整体有序
      .saveAsNewAPIHadoopFile(savePath,
        classOf[ImmutableBytesWritable],
        classOf[KeyValue],
        classOf[HFileOutputFormat2],
        conf)
  }

  /**
   * HFile 导入HBase
   */
  def loadHFileToHbase() = {
    //开始即那个HFile导入到Hbase,此处都是hbase的api操作
    val load: LoadIncrementalHFiles = new LoadIncrementalHFiles(conf)

    //创建hbase的链接,利用默认的配置文件,实际上读取的hbase的master地址
    val conn: Connection = ConnectionFactory.createConnection(conf)

    //根据表名获取表
    val table: Table = conn.getTable(TableName.valueOf(outPutTable))

    //获取hbase表的region分布
    val regionLocator: RegionLocator = conn.getRegionLocator(TableName.valueOf(outPutTable))

    //创建一个hadoop的mapreduce的job
    val job: Job = Job.getInstance(conf)

    //设置job名称
    job.setJobName(s"$outPutTable LoadIncrementalHFiles")

    //此处最重要,需要设置文件输出的key,因为我们要生成HFil,所以outkey要用ImmutableBytesWritable
    job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])

    //输出文件的内容KeyValue
    job.setMapOutputValueClass(classOf[KeyValue])

    //配置HFileOutputFormat2的信息
    HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator)

    //开始导入
    load.doBulkLoad(new Path(savePath), conn.getAdmin, table, regionLocator)
    spark.stop()
  }

  /**
   * 输出初始化结果
   */
  override def printf(): Unit = {
  }

  /**
   * 删除hdfs下的文件
   *
   * @param url 需要删除的路径
   */
  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)
    }
  }
}


打包及执行命令

执行命令:

spark-submit \
--master yarn-client \
--driver-memory 2G \
--executor-memory 4G \
--executor-cores 4 \
--num-executors 4 \
--conf spark.yarn.executor.memoryOverhead=8192 \
--class com.aaa.test.TestHbase \
/home/apps/lillcol/TestHbase.jar \

注:已有Hbase表“outPutTable”,想要查看hbase数据除了hbase shell 还可以关联hive表,
参考:Spark:DataFrame批量导入Hbase的两种方式(HFile、Hive)


异常和错误

非法循环引用

scala.reflect.internal.Symbols$CyclicReference: illegal cyclic reference

Exception in thread "main" scala.reflect.internal.Symbols$CyclicReference: illegal cyclic reference involving object InterfaceAudience
	at scala.reflect.internal.Symbols$Symbol$$anonfun$info$3.apply(Symbols.scala:1502)
	at scala.reflect.internal.Symbols$Symbol$$anonfun$info$3.apply(Symbols.scala:1500)
	at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
	at scala.reflect.internal.Symbols$Symbol.lock(Symbols.scala:546)
	at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1500)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
	at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.info(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$coreLookup$1(JavaMirrors.scala:992)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$lookupClass$1(JavaMirrors.scala:998)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$classToScala1(JavaMirrors.scala:1003)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$classToScala$1.apply(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$classToScala$1.apply(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$toScala$1.apply(JavaMirrors.scala:97)
	at scala.reflect.runtime.TwoWayCaches$TwoWayCache$$anonfun$toScala$1.apply(TwoWayCaches.scala:38)
	at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
	at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
	at scala.reflect.runtime.TwoWayCaches$TwoWayCache.toScala(TwoWayCaches.scala:33)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.toScala(JavaMirrors.scala:95)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.classToScala(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy.<init>(JavaMirrors.scala:163)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy$.apply(JavaMirrors.scala:162)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy$.apply(JavaMirrors.scala:162)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$copyAnnotations(JavaMirrors.scala:683)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$FromJavaClassCompleter.load(JavaMirrors.scala:733)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$FromJavaClassCompleter.complete(JavaMirrors.scala:744)
	at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
	at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$2.info(SynchronizedSymbols.scala:171)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$coreLookup$1(JavaMirrors.scala:992)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$lookupClass$1(JavaMirrors.scala:998)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$classToScala1(JavaMirrors.scala:1003)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$classToScala$1.apply(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$classToScala$1.apply(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$$anonfun$toScala$1.apply(JavaMirrors.scala:97)
	at scala.reflect.runtime.TwoWayCaches$TwoWayCache$$anonfun$toScala$1.apply(TwoWayCaches.scala:38)
	at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
	at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
	at scala.reflect.runtime.TwoWayCaches$TwoWayCache.toScala(TwoWayCaches.scala:33)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.toScala(JavaMirrors.scala:95)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.classToScala(JavaMirrors.scala:980)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy.<init>(JavaMirrors.scala:163)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy$.apply(JavaMirrors.scala:162)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaAnnotationProxy$.apply(JavaMirrors.scala:162)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
	at scala.reflect.runtime.JavaMirrors$JavaMirror.scala$reflect$runtime$JavaMirrors$JavaMirror$$copyAnnotations(JavaMirrors.scala:683)
	at scala.reflect.runtime.JavaMirrors$JavaMirror$FromJavaClassCompleter.load(JavaMirrors.scala:733)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$typeParams$1.apply(SynchronizedSymbols.scala:142)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$typeParams$1.apply(SynchronizedSymbols.scala:133)
	at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
	at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$8.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:168)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.typeParams(SynchronizedSymbols.scala:132)
	at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$8.typeParams(SynchronizedSymbols.scala:168)
	at scala.reflect.internal.Types$NoArgsTypeRef.typeParams(Types.scala:1926)
	at scala.reflect.internal.Types$NoArgsTypeRef.isHigherKinded(Types.scala:1925)
	at scala.reflect.internal.transform.UnCurry$class.scala$reflect$internal$transform$UnCurry$$expandAlias(UnCurry.scala:22)
	at scala.reflect.internal.transform.UnCurry$$anon$2.apply(UnCurry.scala:26)
	at scala.reflect.internal.transform.UnCurry$$anon$2.apply(UnCurry.scala:24)
	at scala.collection.immutable.List.loop$1(List.scala:173)
	at scala.collection.immutable.List.mapConserve(List.scala:189)
	at scala.reflect.internal.tpe.TypeMaps$TypeMap.mapOver(TypeMaps.scala:115)
	at scala.reflect.internal.transform.UnCurry$$anon$2.apply(UnCurry.scala:46)
	at scala.reflect.internal.transform.Transforms$class.transformedType(Transforms.scala:43)
	at scala.reflect.internal.SymbolTable.transformedType(SymbolTable.scala:16)
	at scala.reflect.internal.Types$TypeApiImpl.erasure(Types.scala:225)
	at scala.reflect.internal.Types$TypeApiImpl.erasure(Types.scala:218)
	at org.apache.spark.sql.catalyst.ScalaReflection$class.getClassNameFromType(ScalaReflection.scala:853)
	at org.apache.spark.sql.catalyst.ScalaReflection$.getClassNameFromType(ScalaReflection.scala:39)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$dataTypeFor$1.apply(ScalaReflection.scala:78)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$dataTypeFor$1.apply(ScalaReflection.scala:65)
	at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
	at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:824)
	at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:39)
	at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$dataTypeFor(ScalaReflection.scala:64)
	at org.apache.spark.sql.catalyst.ScalaReflection$.dataTypeFor(ScalaReflection.scala:62)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:63)
	at org.apache.spark.sql.Encoders$.product(Encoders.scala:275)
	at org.apache.spark.sql.LowPrioritySQLImplicits$class.newProductEncoder(SQLImplicits.scala:248)
	at org.apache.spark.sql.SQLImplicits.newProductEncoder(SQLImplicits.scala:34)
	at com.aaa.TestHbase$.main(TestHbase.scala:40)
	at com.aaa.TestHbase.main(TestHbase.scala)

这个错误的意思是非法的循环引用
目前我没搞明白我循环引用了啥,不过大概摸清了出现异常的情况。
异常出现的代码块:

val result1 : RDD[(ImmutableBytesWritable, Seq[KeyValue])] = TM_D
      .map(row => {
        var kvlist: Seq[KeyValue] = List()
        var kv: KeyValue = null
        val cf: Array[Byte] = clounmFamily.getBytes //列族
        val rowKey = Bytes.toBytes(row.getAs[Int]("ID"))
        val immutableRowKey = new ImmutableBytesWritable(rowKey)
        for (i <- 1 to (columnsName.length - 1)) {
          //          将rdd转换成HFile需要的格式,
          //          我们上面定义了Hfile的key是ImmutableBytesWritable,
          //          那么我们定义的RDD也是要以ImmutableBytesWritable的实例为key
          kv = new KeyValue(rowKey, cf, Bytes.toBytes(columnsName(i)), Bytes.toBytes(row.get(i) + ""))
          //          将新的kv加在kvlist后面(不能反 需要整体有序)
          kvlist = kvlist :+ kv
        }
        //(rowKey, kvlist.length)-----1
        //(rowKey, kvlist)-----2
        //(immutableRowKey, kvlist.length)-----3
        //(immutableRowKey, kvlist)-----4
      })

如上面的代码所示:
如果最后的返回值是2、3、4中的一个,那么就会报这个非法循环引用的错误
他们的共同点是都是对象(虽然scala万物皆可对象,但是还是没搞懂);
如果返回的是1则没有问题,但是这并不是我们要的答案。

网上一堆说scala版本问题,JDK版本问题,广播变量等都没有解决,只能自己慢慢捣鼓。

通过观察数据类型发现TM_D是DataFrame/Dataset[Row]
进行map操作后还是DataFrame/Dataset[Row],但是编译期间没有报错;
有可能因为是DataFrame/Dataset[Row]map操作有我不知道的限制,所以果断DataFrame/Dataset[Row]转RDD试试。
嗯......~,还真的给我试出来了,运气成分,我现在也不知道啥原因,也许是天选之子吧。
关于转换的才做可以参考我的博客Spark 读写数据、抽象转换 拾遗
修改后的代码(未优化):

   val result1: RDD[(ImmutableBytesWritable, Seq[KeyValue])] = TM_D
      .rdd  //转换rdd
      .map(row => {
        var kvlist: Seq[KeyValue] = List()
        var kv: KeyValue = null
        val cf: Array[Byte] = clounmFamily.getBytes //列族
        val rowKey = Bytes.toBytes(row.getAs[Int]("ID"))
        val immutableRowKey = new ImmutableBytesWritable(rowKey)
        for (i <- 1 to (columnsName.length - 1)) {
          kv = new KeyValue(rowKey, cf, Bytes.toBytes(columnsName(i)), Bytes.toBytes(row.get(i) + ""))
          kvlist = kvlist :+ kv
        }
        (immutableRowKey, kvlist)
      })

key排序

Added a key not lexically larger than previous

Caused by: java.io.IOException: Added a key not lexically larger than previous. Current cell = \x00\x00\xE4h/cf:CN_TAG/1568255140650/Put/vlen=3/seqid=0, lastCell = \x00\x00\xE4h/cf:FIRST_DT/1568255140650/Put/vlen=6/seqid=0
        at org.apache.hadoop.hbase.io.hfile.HFileWriterImpl.checkKey(HFileWriterImpl.java:245)
        at org.apache.hadoop.hbase.io.hfile.HFileWriterImpl.append(HFileWriterImpl.java:731)
        at org.apache.hadoop.hbase.regionserver.StoreFileWriter.append(StoreFileWriter.java:234)
        at org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2$1.write(HFileOutputFormat2.java:344)
        at org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2$1.write(HFileOutputFormat2.java:231)
        at org.apache.spark.internal.io.HadoopMapReduceWriteConfigUtil.write(SparkHadoopWriter.scala:356)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:130)
        at org.apache.spark.internal.io.SparkHadoopWriter$$anonfun$4.apply(SparkHadoopWriter.scala:127)
        at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1415)
        at org.apache.spark.internal.io.SparkHadoopWriter$.org$apache$spark$internal$io$SparkHadoopWriter$$executeTask(SparkHadoopWriter.scala:139)

Hbase查询是根据rowkey进行查询的,并且rowkey是有序,某种程度上来说rowkey就是一个索引,这是Hbase查询高效的一个原因。
一开始代码中只是对key排序,在旧的版本测试没问题,但是2.0.2出问题了。
此处报错的意思是当前列CN_TAG 比 上一列FIRST_DT小,
猜测同一个key下clounm也需要有序,
于是对key,clounm排序解决了这个问题。
(之前的博客中应该是因为一开始对列排了序 所以没出问题)。

解决方法:

 hfileRDD
      .sortBy(x => (x._1, x._2.getKeyString), true) //要保持 整体有序
      .saveAsNewAPIHadoopFile(savePath,
        classOf[ImmutableBytesWritable],
        classOf[KeyValue],
        classOf[HFileOutputFormat2],
        conf)

HBase 根目录不存在

java.util.concurrent.ExecutionException: org.apache.phoenix.shaded.org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /hbase/hbaseid
        at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
        at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
        at org.apache.hadoop.hbase.client.ConnectionImplementation.retrieveClusterId(ConnectionImplementation.java:549)
        at org.apache.hadoop.hbase.client.ConnectionImplementation.<init>(ConnectionImplementation.java:287)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:219)
        at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:114)
        at com.aaa.TestHbase$.main(TestHbase.scala:99)
        at com.aaa.TestHbase.main(TestHbase.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:904)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:198)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:228)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:137)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

默认为:/hbase
如果修改了需要指定,否则找不到该路径

修改方式有两个:


  • 修改配置文件bhase-site.xml
<configuration>
  <property>
    <name>hbase.unsafe.stream.capability.enforce</name>
    <value>false</value>
  </property>

  <property>
    <name>zookeeper.znode.parent</name>
    <value>/hbase</value>
  </property>
</configuration>
  • 代码中设置参数
    代码中执行要使用此方法
conf.set("hbase.unsafe.stream.capability.enforce", "false") //hbase  根目录设定
conf.set("zookeeper.znode.parent", "/hbase") //设置成真实的值

一个family下超过了默认的32个hfile

Exception in thread "main" java.io.IOException: Trying to load more than 32 hfiles to one family of one region
        at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles.doBulkLoad(LoadIncrementalHFiles.java:288)
        at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles.run(LoadIncrementalHFiles.java:842)
        at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)
        at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:84)
        at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles.main(LoadIncrementalHFiles.java:847)

解决办法有两个:

  • 修改配置文件bhase-site.xml
  <property>
    <name>hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily</name>
    <value>400</value>
  </property>
  • 代码中设置参数
    代码中执行要使用此方法
conf.set("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", "400")

内存溢出

19/09/17 15:25:17 ERROR YarnScheduler: 
Lost executor 8 on slave2: Container killed by YARN for exceeding memory limits. 
11.0 GB of 11 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

excutor 内存不够,这个就要根据自己的代码进行调整了,
加大内存总量不一定有用,也不存在万能的方法,但是可以根据下面的思路去尝试。

  1. spark.yarn.executor.memoryOverhead设置为最大值,可以考虑一下4096。这个数值一般都是2的次幂。
  2. 加大rdd、DataFrame分区,像我repartition(200),前提是数据是均匀分布的,否则可能会出现数据倾斜。
  3. 减少将spark.executor.core如:从8设置为4。将core的个数调小。
  4. 增加将spark.executor.memory如:从8g设置为12g。将内存调大。
  • spark.yarn.executor.memoryOverhead计算方式
E = max(MEMORY_OVERHEAD_MIN,MEMORY_OVERHEAD_FACTOR*executorMemory)

MEMORY_OVERHEAD_FACTOR默认为0.1;
executorMemory为设置的executor-memory;
MEMORY_OVERHEAD_MIN默认为384m;
参数MEMORY_OVERHEAD_FACTOR和MEMORY_OVERHEAD_MIN一般不能直接修改,是Spark代码中直接写死的

  • executor可用内存的计算方式:
E = (driver-memory+spark.yarn.executor.memoryOverhead)

本文为原创文章,转载请注明出处!!!

分类:

技术点:

相关文章: