分享

Spark 批量写数据入HBase

yuwenge 2015-6-2 19:57:21 发表于 小知识点 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 4 19840



介绍

  工作中常常会遇到这种情形,需要将hdfs中的大批量数据导入HBase。本文使用Spark+HBase的方式将RDD中的数据导入HBase中。没有使用官网提供的newAPIHadoopRDD接口的方式。使用本文的方式将数据导入HBase, 7000W条数据,花费时间大概20分钟左右,本文Spark可用核数量为20。

本文使用spark版本为1.3.0,hbase版本为0.98.1

hbase表结构为:表名table,列族Family,列为qualifier。

代码如下:

[mw_shl_code=scala,true]val readFile = sc.textFile("/path/to/file").map(x => x.split(","))
val tableName = "table"
readFile.foreachPartition{
  x=> {
    val myConf = HBaseConfiguration.create()
    myConf.set("hbase.zookeeper.quorum", "web102,web101,web100")
    myConf.set("hbase.zookeeper.property.clientPort", "2181")
    myConf.set("hbase.defaults.for.version.skip", "true")
    val myTable = new HTable(myConf, TableName.valueOf(tableName))
    myTable.setAutoFlush(false, false)//关键点1
    myTable.setWriteBufferSize(3*1024*1024)//关键点2
    x.foreach { y => {
      println(y(0) + ":::" + y(1))
      val p = new Put(Bytes.toBytes(y(0)))
      p.add("Family".getBytes, "qualifier".getBytes, Bytes.toBytes(y(1)))
      myTable.put(p)
    }
    }
    myTable.flushCommits()//关键点3
  }
}[/mw_shl_code]


此程序是使用了RDD的foreachPartition函数,在此程序中有三个比较关键的地方。
关键点1_:将自动提交关闭,如果不关闭,每写一条数据都会进行提交,是导入数据较慢的做主要因素。
关键点2:设置缓存大小,当缓存大于设置值时,hbase会自动提交。此处可自己尝试大小,一般对大数据量,设置为5M即可,本文设置为3M。
关键点3:每一个分片结束后都进行flushCommits(),如果不执行,当hbase最后缓存小于上面设定值时,不会进行提交,导致数据丢失。

注:此外如果想提高Spark写数据如Hbase速度,可以增加Spark可用核数量。



来源:mach_learn





已有(4)人评论

跳转到指定楼层
小南3707 发表于 2015-6-3 09:10:43
赞!              
回复

使用道具 举报

haorengoodman 发表于 2015-6-3 10:18:03
new PairRDDFunctions[ImmutableBytesWritable, Put](xxx).saveAsNewAPIHadoopDataset(jobConfiguration())
回复

使用道具 举报

haorengoodman 发表于 2015-6-3 10:21:44
想请问一下楼主,在执行spark任务时,使用了多大的资源量
回复

使用道具 举报

sprite101 发表于 2015-6-3 12:52:38
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条