SparkStreaming 写数据到 HBase,由于共用连接造成的数据丢失问题

有如下程序,SparkStreaming 读取 Kafka 中的数据,经过处理后,把数据写入到 Hbase 中

/**
  * Author: Jed
  * Description: SparkStreaming 读取 Kafka 中的数据,实时写入 HBase中
  * Create: 2018-05-04 14:50
  */
object HBaseTest {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(2))

    val kafkaParams = Map[String, AnyRef](
      "bootstrap.servers" -> "172.16.26.6:9092,172.16.26.10:9092,172.16.26.13:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest"
      "group.id" -> s"GROUP${new Random().nextInt(1000)}"
    )

    val topics = Array("baihe")

    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val values: DStream[Array[String]] = stream.map(_.value.split("\\|"))

    values.foreachRDD(rdd => {
      rdd.foreachPartition(partition => {
        val connection = HBaseUtil.getConnection
        val tableName = TableName.valueOf("test")
        val table = connection.getTable(tableName)
        val puts = new ArrayList[Put]

        try {
          partition.foreach(arr => {

            val put = new Put(CustomerFunction.genRowkey(arr(0)))
            val index = Array[Int](0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            val enNames = Array[String]("touched", "user_number", "start_time", "end_time", "channel_id", "binding_flag", "click_path", "result", "interf_name", "channel_category", "by_operator")
            var value = ""
            for (i <- 0 until index.length) {
              value += arr(i) + "|"
            }
            value = value.dropRight(1)
            put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("q"), Bytes.toBytes(value))
            puts.add(put)
            // 这里为了提高性能,每一万条入一次HBase库
            if (puts.size % 10000 == 0) {
              table.put(puts)
              puts.clear()
            }
          })
        } catch {
          case e: Exception => e.printStackTrace
        } finally {
          table.put(puts)
          table.close
          connection.close
        }
      })
    })

    ssc.start
    ssc.awaitTermination
  }
}


object HBaseUtil {

  var conf: Configuration = null
  var connection: Connection = null

  def getConnection(): Connection = {

    if (conf == null) {
      conf.set("hbase.zookeeper.quorum", "172.16.26.6:2181,172.16.26.10:2181,172.16.26.13:2181")
    }

    if ((connection == null || connection.isClosed()) && conf != null) {
      try {
        connection = ConnectionFactory.createConnection(conf)
      } catch {
        case e: Exception => e.printStackTrace()
      }
    }
    return connection;
  }

  def colse() = {
    if (connection != null) {
      try {
        connection.close();
      } catch {
        case e: Exception => e.printStackTrace()
      }
    }
  }
}

执行以上程序,中途会报错:

2018-05-29 16:21:40 883 [ERROR] org.apache.hadoop.hbase.client.AsyncProcess.submit(AsyncProcess.java:432) Failed to get region location 
org.apache.hadoop.hbase.DoNotRetryIOException: hconnection-0x6432ad81 closed
    at org.apache.hadoop.hbase.client.ConnectionManager$HConnectionImplementation.locateRegion(ConnectionManager.java:1174)
    at org.apache.hadoop.hbase.client.AsyncProcess.submit(AsyncProcess.java:422)
    at org.apache.hadoop.hbase.client.AsyncProcess.submit(AsyncProcess.java:371)
    at org.apache.hadoop.hbase.client.BufferedMutatorImpl.backgroundFlushCommits(BufferedMutatorImpl.java:245)
    at org.apache.hadoop.hbase.client.BufferedMutatorImpl.flush(BufferedMutatorImpl.java:197)
    at org.apache.hadoop.hbase.client.HTable.flushCommits(HTable.java:1461)
    at org.apache.hadoop.hbase.client.HTable.put(HTable.java:1029)

重点是:hconnection-0x6432ad81 closed 问题出在获得连接的工具类中,在 DStream 中的每个 partition 中获得中一个 HBase 的连接,为了提高"效率",让每个 partition 共用了一个 connection,但就是这样,才导致了问题的出现,假设 A partition 中有 10000 条数据,B partition 中有 20000 条数据,两个 partition 共用一个 connection,A、B两个 partition 并行的往 HBase 中写数据,当 A partition 写完10000条数据后,关闭了 connection,假设此时 B partition 也已经写入了10000条数据,但它还有 10000 条数据要写,连接却关闭了,程序会报以上的错误,数据会丢失 10000 条

解决办法就是让每个 partition 获得独立的 connection,只需要把 HBaseUtil 类修改如下即可:

object HBaseUtil {
  val conf: Configuration = HBaseConfiguration.create()
  conf.set("hbase.zookeeper.quorum", "192.168.42.101:2181,192.168.42.102:2181,192.168.42.101:2181")
  def getConnection(): Connection = {
    return ConnectionFactory.createConnection(conf)
  }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏伦少的博客

spark基本概念(便于自己随时查阅--摘自Spark快速大数据分析)

转载请务必注明原创地址为:http://dongkelun.com/2018/01/23/sparkBasicConcept/

3888
来自专栏Small Code

【Python】自动生成命令行工具 - fire 简介

Python 中用于生成命令行接口(Command Line Interfaces, CLIs)的工具已经有一些了,例如已经成为 Python 标准库的 arg...

4829
来自专栏张浩的专栏

Hive 创建自定义函数(UDF)

当Hive中的内置函数不满足我们需求的时候,我们可以自定义我们自己的Hive函数,来满足我们的需求。

4091
来自专栏JackeyGao的博客

Django小技巧22: 设计一个好的模型

本篇将分享一些技巧,用户改进 Model 的设计。其中有很多与命名约定有关, 这可以大大的提高代码的可读性。

1622
来自专栏祝威廉

Structured Streaming如何实现Parquet存储目录按时间分区

StreamingPro现在支持以SQL脚本的形式写Structured Streaming流式程序了: mlsql-stream。不过期间遇到个问题,我希望按...

1501
来自专栏Jed的技术阶梯

Kafka 自定义分区器

(1) 如果键值为 null,并且使用了默认的分区器,那么记录将被随机地发送到主题内各个可用的分区上。分区器使用轮询(Round Robin)算法将消息均衡地分...

1792
来自专栏null的专栏

Hive——巧用transform处理复杂的字符串问题

相比于Map-Reduce,Hive对数据的处理相对简单,但是Hive本身提供的函数,对于处理复杂的字符串问题,就显得不是很方便,此时,可以借助transfor...

4405
来自专栏祝威廉

利用 Spark DataSource API 实现Rest数据源

先说下这个需求的来源。通常在一个流式计算的主流程里,会用到很多映射数据,譬如某某对照关系,而这些映射数据通常是通过HTTP接口暴露出来的,尤其是外部系统,你基本...

1772
来自专栏岑玉海

hbase源码系列(十二)Get、Scan在服务端是如何处理?

继上一篇讲了Put和Delete之后,这一篇我们讲Get和Scan, 因为我发现这两个操作几乎是一样的过程,就像之前的Put和Delete一样,上一篇我本来只打...

56110
来自专栏牛肉圆粉不加葱

[Spark源码剖析]Pool-Standalone模式下的队列Pool-Spark Standalone模式下的队列

org.apache.spark.scheduler.Pool是 Spark Standalone 模式下的队列。从其重要成员及成员函数来剖析这个在 TaskS...

611

扫码关注云+社区

领取腾讯云代金券