背景
之前的博客：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 內存不夠，這個就要根據自己的代碼進行調整了，
加大內存總量不一定有用，也不存在萬能的方法，但是可以根據下面的思路去嘗試。
- 將
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)
本文為原創文章，轉載請注明出處！！！