about云开发

 找回密码
 立即注册

QQ登录

只需一步,快速开始

扫一扫,访问微社区

查看: 532|回复: 1
打印 上一主题 下一主题

[总结型] 看看大神是如何解决难题的:spark写hfile遇到错误解题思路总结

[复制链接]

464

主题

400

听众

243

收听

版主

Rank: 7Rank: 7Rank: 7

积分
5510
跳转到指定楼层
楼主
发表于 2019-5-14 19:41:45 | 只看该作者 回帖奖励 |倒序浏览 |阅读模式
问题导读

1.本文解决了什么问题?
2.在写入hbase的过程中遇到了什么问题?
3.本文解决问题的思路是什么?


需求
将HDFS上的数据解析出来,然后通过hfile方式批量量写入Hbase(需要多列列写入)
写入数据的关键api:
[Scala] 纯文本查看 复制代码
saveAsNewAPIHadoopFile(
stagingFolder,
classOf[ImmutableBytesWritable],
classOf[KeyValue],
classOf[HFileOutputFormat2],
job.getConfiguration)




特殊地方:
1):
最初写hfile警告:
Does it contain files in subdirectories that correspond to column family
names

这个原因大概3种:
A:代码问题
B:数据源问题
C:setMapOutputKeyClass 和 saveAsNewAPIHadoopFile中的Class不不⼀一致
(我的是数据源问题)

2):
正常些put操作的时候,服务端⾃自动帮助排序,因此在使用put操作的时候没有涉及到这样的错误:


Added a key not lexically larger than previous

但是在写hfile的时候如果出现报错:
Added a key not lexically larger than previous

这样的错误,一般会认为rowkey没有做好排序,然后傻fufu的去验证了了一下,rowkey的确做了了排序
真正原因:
spark写hfile时候是按照rowkey+列列族+列列名进行行排序的,因此在写入数据的时候,要做到整体有序
(事情还没完)

3):
因为需要多列列写入,最好的方式:要么反射来动态获取列列名称和列列值 、 要么通过datafame去获取
(df.columns)
反射方式:
[Scala] 纯文本查看 复制代码
val listData: RDD[(ImmutableBytesWritable, ListBuffer[KeyValue])] = rdd.map
{
line =>
val rowkey = line.vintime
val clazz = Class.forName(XXXXXXXXXXXXXXXX)
val fields = clazz.getDeclaredFields
var list = new ListBuffer[String]()
var kvlist = new ListBuffer[KeyValue]()//
if (fields != null && fields.size > 0) {
for (field <- fields) {
field.setAccessible(true)
val column = field.getName
list.append(column)
}
}
val newList = list.sortWith(_ < _)
val ik = new ImmutableBytesWritable(Bytes.toBytes(rowkey))
for(column <- newList){
val declaredField: Field =
line.getClass.getDeclaredField(column)
declaredField.setAccessible(true)
val value = declaredField.get(line).toString
val kv: KeyValue = new KeyValue(
Bytes.toBytes(rowkey),
Bytes.toBytes(columnFamily),
Bytes.toBytes(column),
Bytes.toBytes(value))
kvlist.append(kv)
}
(ik, kvlist)
}



datafame方式:
[Scala] 纯文本查看 复制代码
val tmpData: RDD[(ImmutableBytesWritable, util.LinkedList[KeyValue])] =
df.rdd.map(
line =>{
val rowkey = line.getAs[String]("vintime")
val ik = new ImmutableBytesWritable(Bytes.toBytes(rowkey))
var linkedList = new util.LinkedList[KeyValue]()
for (column <- columns) {
val kv: KeyValue = new KeyValue(
Bytes.toBytes(rowkey),
Bytes.toBytes(columnFamily),
Bytes.toBytes(column),
Bytes.toBytes(line.getAs[String](column)))
linkedList.add(kv)
}
(ik, linkedList)
})
val result: RDD[(ImmutableBytesWritable, KeyValue)] =
tmpData.flatMapValues(
s => {
val values: Iterator[KeyValue] =
JavaConverters.asScalaIteratorConverter(s.iterator()).asScala
values
}
).sortBy(x =>x._1 , true)


仔细观察可以发现,其实两者都做了了排序操作,但是即便便经过(1)步骤后仍然报错:
Added a key not lexically larger than previous

那么在回想一下之前写hfile的要求:
rowkey+列列族+列列都要有序,那么如果出现数据的重复,也不不算是有序的操作!
因为,做一下数据的去重:
[Scala] 纯文本查看 复制代码
val key: RDD[(String, TransferTime)] = data.reduceByKey((x, y) => y)
val unitData: RDD[TransferTime] = key.map(line => line._2)

果然,这样解决了了:Added a key not lexically larger than previous这个异常
但是会报如下另一个异常:
[Scala] 纯文本查看 复制代码
Kryo serialization failed: Buffer overflow

这个是因为在对一些类做kryo序列列化时候,数据量量的缓存大小超过了了默认值,做一下调整即可
[Scala] 纯文本查看 复制代码
sparkConf.set("spark.kryoserializer.buffer.max" , "256m")
sparkConf.set("spark.kryoserializer.buffer" , "64m")

完整代码:
[Scala] 纯文本查看 复制代码
/**
* Created by angel
*/
object WriteTransferTime extends WriteToHbase{
/**
* @param data 要插⼊入的数据
* @param tableName 表名
**/
override def bulkLoadData(data: RDD[Any], tableName: String ,
columnFamily:String): Unit = {
val bean: RDD[TransferTime] = data.map(line =>
line.asInstanceOf[TransferTime])
val map: RDD[(String, TransferTime)] = bean.map(line => (line.vintime ,
line))
val key: RDD[(String, TransferTime)] = map.reduceByKey((x, y) => y)
val map1: RDD[TransferTime] = key.map(line => line._2)
val by1: RDD[TransferTime] = map1.sortBy(f => f.vintime)
val listData: RDD[(ImmutableBytesWritable, ListBuffer[KeyValue])] =
by1.map {
line =>
val rowkey = line.vintime
val clazz =
Class.forName("com.dongfeng.code.Bean.message.TransferTime")
val fields = clazz.getDeclaredFields
var list = new ListBuffer[String]()
var kvlist = new ListBuffer[KeyValue]()//
if (fields != null && fields.size > 0) {
for (field <- fields) {
field.setAccessible(true)
val column = field.getName
list.append(column)
}
}
val newList = list.sortWith(_ < _)
val ik = new ImmutableBytesWritable(Bytes.toBytes(rowkey))
for(column <- newList){
val declaredField: Field = line.getClass.getDeclaredField(column)
declaredField.setAccessible(true)
val value = declaredField.get(line).toString
val kv: KeyValue = new KeyValue(
Bytes.toBytes(rowkey),
Bytes.toBytes(columnFamily),
Bytes.toBytes(column),
Bytes.toBytes(value))
kvlist.append(kv)
}
(ik, kvlist)
}
val result: RDD[(ImmutableBytesWritable, KeyValue)] =
listData.flatMapValues(
s => {
val values: Iterator[KeyValue] = s.iterator
values
}
)
val resultDD: RDD[(ImmutableBytesWritable, KeyValue)] = result.sortBy(x
=>x._1 , true)
WriteToHbaseDB.hfile_load(result , TableName.valueOf(tableName) ,
columnFamily)
}
}
def hfile_load(rdd:RDD[(ImmutableBytesWritable , KeyValue)] , tableName:
TableName , columnFamily:String): Unit ={
//声明表的信息
var table: Table = null
try{
val startTime = System.currentTimeMillis()
println(s"开始时间:-------->${startTime}")
//⽣生成的HFile的临时保存路路径
val stagingFolder = "hdfs://cdh1:9000/hfile/"+tableName+new
Date().getTime//
table = connection.getTable(tableName)
//如果表不不存在,则创建表
if(!admin.tableExists(tableName)){
createTable(tableName , columnFamily)
}
//开始导⼊入
val job = Job.getInstance(config)
job.setJobName("DumpFile")
job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
job.setMapOutputValueClass(classOf[KeyValue])
rdd.sortBy(x => x._1, true).saveAsNewAPIHadoopFile(
stagingFolder,
classOf[ImmutableBytesWritable],
classOf[KeyValue],
classOf[HFileOutputFormat2],
job.getConfiguration)
val load = new LoadIncrementalHFiles(config)
val regionLocator = connection.getRegionLocator(tableName)
HFileOutputFormat2.configureIncrementalLoad(job, table,
regionLocator)
load.doBulkLoad(new Path(stagingFolder), table.asInstanceOf[HTable])
// load.doBulkLoad(new Path(stagingFolder) , connection.getAdmin ,
table , regionLocator)
val endTime = System.currentTimeMillis()
println(s"结束时间:-------->${endTime}")
println(s"花费的时间:----------------->${(endTime - startTime)}ms")
}catch{
case e:IOException =>
e.printStackTrace()
}finally {
if (table != null) {
try {
// 关闭HTable对象 table.close();
} catch {
case e: IOException =>
e.printStackTrace();
}
}
if (connection != null) {
try { //关闭hbase连接.
connection.close();
} catch {
case e: IOException =>
e.printStackTrace();
}
}
}
}
}



来自文档-下载:
spark写hfile遇到错误以及总结.zip (107.62 KB, 下载次数: 2)

最新经典文章,欢迎关注公众号
分享到:  QQ好友和群QQ好友和群 QQ空间QQ空间 腾讯微博腾讯微博 腾讯朋友腾讯朋友 微信微信
收藏收藏 转播转播 分享分享 分享淘帖 赞 踩 分享到微信分享到微信
欢迎加入about云群425860289432264021 ,云计算爱好者群,关注about云腾讯认证空间
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /3 下一条

QQ|小黑屋|大发快3-首页 ( 京ICP备12023829号

GMT+8, 2019-6-18 06:38 , Processed in 0.395365 second(s), 35 queries , Gzip On.

Powered by Discuz! X3.2 Licensed

快速回复 返回顶部 返回列表