背景
之前的博客：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_ODR_BANK_STAT_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_ODR_BANK_STAT_D是DataFrame/Dataset[Row]，
进行map操作后还是DataFrame/Dataset[Row],但是编译期间没有报错；
有可能因为是DataFrame/Dataset[Row]的 map操作有我不知道的限制，所以果断DataFrame/Dataset[Row]转RDD试试。
嗯......~，还真的给我试出来了，运气成分，我现在也不知道啥原因，也许是天选之子吧。
关于转换的操作可以参考我的博客Spark 读写数据、抽象转换 拾遗
修改后的代码(未优化)：
val result1: RDD[(ImmutableBytesWritable, Seq[KeyValue])] = TM_ODR_BANK_STAT_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:CNSM_CNT_TAG/1568255140650/Put/vlen=3/seqid=0, lastCell = \x00\x00\xE4h/cf:FIRST_ACTV_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出问题了。
此处报错的意思是当前列CNSM_CNT_TAG 比 上一列FIRST_ACTV_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 内存不够，这个就要根据自己的代码进行调整了，
加大内存总量不一定有用，也不存在万能的方法，但是可以根据下面的思路去尝试。
- 将
spark.yarn.executor.memoryOverhead设置为最大值，可以考虑一下4096。这个数值一般都是2的次幂。 - 加大rdd、DataFrame分区，像我repartition(200)，前提是数据是均匀分布的，否则可能会出现数据倾斜。
- 减少将
spark.executor.core如：从8设置为4。将core的个数调小。 - 增加将
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)
本文为原创文章，转载请注明出处！！！
网友评论