Spark写入HBase（Bulk方式）

2024年1月26日 91次阅读 来源: spark

在使用Spark时经常需要把数据落入HBase中，如果使用普通的Java API，写入会速度很慢。还好Spark提供了Bulk写入方式的接口。那么Bulk写入与普通写入相比有什么优势呢？

  • BulkLoad不会写WAL，也不会产生flush以及split。
  • 如果我们大量调用PUT接口插入数据，可能会导致大量的GC操作。除了影响性能之外，严重时甚至可能会对HBase节点的稳定性造成影响。但是采用Bulk就不会有这个顾虑。
  • 过程中没有大量的接口调用消耗性能

下面给出完整代码：

import org.apache.hadoop.hbase.client.{Put, Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.conf.Configuration

/**
  * Created by shaonian
  */
object HBaseBulk {

  def main(args: Array[String]): Unit = {
　　val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Bulk")
　　val sc = new SparkContext(sparkConf)

　　val conf = new Configuration()
　　conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3")
　　conf.set("hbase.zookeeper.property.clientPort", "2181")
　　conf.set(TableOutputFormat.OUTPUT_TABLE, "bulktest")
　　val job = Job.getInstance(conf)
　　job.setOutputKeyClass(classOf[ImmutableBytesWritable])
　　job.setOutputValueClass(classOf[Result])
　　job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])

　　val init = sc.makeRDD(Array("1,james,32", "2,lebron,30", "3,harden,28"))
　　val rdd = init.map(_.split(",")).map(arr => {
　　  val put = new Put(Bytes.toBytes(arr(0)))
　　  put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes(arr(1)))
　　  put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"), Bytes.toBytes(arr(2).toInt))
　　  (new ImmutableBytesWritable, put)
　　})
　　rdd.saveAsNewAPIHadoopDataset(job.getConfiguration)
　　sc.stop()
  }
    原文作者：spark
    原文地址: https://www.cnblogs.com/dreamtecher/p/6824378.html
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞