Spark与HBase的整合

前言
之前因为仅仅是把HBase当成一个可横向扩展并且具有持久化能力的KV数据库,所以只用在了指标存储上,参看很早之前的一篇文章基于HBase做Storm 实时计算指标存储。这次将HBase用在了用户行为存储上,因为Rowkey的过滤功能也很不错,可以很方便的把按人或者内容的维度过滤出所有的行为。从某种意义上,HBase的是一个有且仅有一个多字段复合索引的存储引擎。

虽然我比较推崇实时计算,不过补数据或者计算历史数据啥的,批处理还是少不了的。对于历史数据的计算,其实我是有两个选择的,一个是基于HBase的已经存储好的行为数据进行计算,或者基于Hive的原始数据进行计算,最终选择了前者,这就涉及到Spark(StreamingPro) 对HBase的批处理操作了。
整合过程

和Spark 整合,意味着最好能有Schema(Mapping),因为Dataframe 以及SQL API 都要求你有Schema。 遗憾的是HBase 有没有Schema取决于使用者和场景。通常SparkOnHBase的库都要求你定义一个Mapping(Schema),比如hortonworks的 SHC(https://github.com/hortonworks-spark/shc) 就要求你定义一个如下的配置:

{
"rowkey":"key",
"table":{"namespace":"default", "name":"pi_user_log", "tableCoder":"PrimitiveType"},
"columns":{"col0":{"cf":"rowkey", "col":"key", "type":"string"},
"col1":{"cf":"f","col":"col1", "type":"string"}
}
}

看上面的定义已经还是很容易看出来的。对HBase的一个列族和列取一个名字,这样就可以在Spark的DataSource API使用了,关于如何开发Spark DataSource API可以参考我的这篇文章利用 Spark DataSource API 实现Rest数据源中使用,SHC大体实现的就是这个API。现在你可以这么用了:

 val cat = "{\n\"rowkey\":\"key\",\"table\":{\"namespace\":\"default\", \"name\":\"pi_user_log\", \"tableCoder\":\"PrimitiveType\"},\n\"columns\":{\"col0\":{\"cf\":\"rowkey\", \"col\":\"key\", \"type\":\"string\"},\n\"28360592\":{\"cf\":\"f\",\"col\":\"28360592\", \"type\":\"string\"}\n}\n}"
    val cc = sqlContext
      .read
      .options(Map(HBaseTableCatalog.tableCatalog -> cat))
      .format("org.apache.spark.sql.execution.datasources.hbase")
      .load()

不过当你有成千上万个列,那么这个就无解了,你不大可能一一定义,而且很多时候使用者也不知道会有哪些列,列名甚至可能是一个时间戳。我们现在好几种情况都遇到了,所以都需要解决:

  1. 自动获取HBase里所有的列形成Schema,这样就不需要用户配置了。
  2. 规定HBase只有两个列,一个rowkey,一个 content,content 是一个map,包含所有以列族+列名为key,对应内容为value。

先说说第二种方案(因为其实第一种方案也要依赖于第二种方案):

{
        "name": "batch.sources",
        "params": [
          {
            "inputTableName": "log1",
            "format": "org.apache.spark.sql.execution.datasources.hbase.raw",
            "path": "-",
            "outputTable": "log1"
          }
        ]
      },
      {
        "name": "batch.sql",
        "params": [
          {
            "sql": "select rowkey,json_value_collect(content) as actionList from log1",
            "outputTableName":"finalTable"
          }
        ]
      },

首先我们配置了一个HBase的表,叫log1,当然,这里是因为程序通过hbase-site.xml获得HBase的链接,所以配置上你看不到HBase相关的信息。接着呢,在SQL 里你就可以对content 做处理了。我这里是把content 转化成了JSON格式字符串。再之后你就可以自己写一个UDF函数之类的做处理了,从而实现你复杂的业务逻辑。我们其实每个字段里存储的都是JSON,所以我其实不关心列名,只要让我拿到所有的列就好。而上面的例子正好能够满足我这个需求了。

而且实现这个HBase DataSource 也很简单,核心逻辑大体如下:

case class HBaseRelation(
                          parameters: Map[String, String],
                          userSpecifiedschema: Option[StructType]
                        )(@transient val sqlContext: SQLContext)
  extends BaseRelation with TableScan with Logging {

  val hbaseConf = HBaseConfiguration.create()

  def buildScan(): RDD[Row] = {
    hbaseConf.set(TableInputFormat.INPUT_TABLE, parameters("inputTableName"))
    val hBaseRDD = sqlContext.sparkContext.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
      .map { line =>
        val rowKey = Bytes.toString(line._2.getRow)

        import net.liftweb.{json => SJSon}
        implicit val formats = SJSon.Serialization.formats(SJSon.NoTypeHints)

        val content = line._2.getMap.navigableKeySet().flatMap { f =>
          line._2.getFamilyMap(f).map { c =>
            (Bytes.toString(f) + ":" + Bytes.toString(c._1), Bytes.toString(c._2))
          }
        }.toMap

        val contentStr = SJSon.Serialization.write(content)

        Row.fromSeq(Seq(UTF8String.fromString(rowKey), UTF8String.fromString(contentStr)))
      }
    hBaseRDD
  }
}

那么我们回过头来,如何让Spark自动发现Schema呢?大体你还是需要过滤所有数据得到列的合集,然后形成Schema的,成本开销很大。我们也可以先将我们的数据转化为JSON格式,然后就可以利用Spark已经支持的JSON格式来自动推倒Schema的能力了。

总体而言,其实并不太鼓励大家使用Spark 对HBase进行批处理,因为这很容易让HBase过载,比如内存溢出导致RegionServer 挂掉,最遗憾的地方是一旦RegionServer 挂掉了,会有一段时间读写不可用,而HBase 又很容易作为实时在线程序的存储,所以影响很大。

时间: 2024-05-08 14:32:44

Spark与HBase的整合的相关文章

spark往hbase写数据

问题描述 valresult:org.apache.spark.rdd.RDD[(String,Int)]result.foreach(res=>{varput=newPut(java.util.UUID.randomUUID().toString.reverse.getBytes()).add("lv6".getBytes(),res._1.toString.getBytes(),res._2.toString.getBytes)table.put(put)})上面是程序,re

开源大数据技术专场(上午):Spark、HBase、JStorm应用与实践

16日上午9点,2016云栖大会"开源大数据技术专场" (全天)在阿里云技术专家封神的主持下开启.通过封神了解到,在上午的专场中,阿里云高级技术专家无谓.阿里云技术专家封神.阿里巴巴中间件技术部高级技术专家天梧.阿里巴巴中间件技术部资深技术专家纪君祥将给大家带来Hadoop.Spark.HBase.JStorm Turbo等内容. 无谓:Hadoop过去现在未来,从阿里云梯到E-MapReduce 阿里云高级技术专家 无谓  从开辟大数据先河至现在,风雨十年,Hadoop已成为企业的通

spark 调用hbase出现Cannot create a record reader because of a previous error

问题描述 使用spark调用Hbase时出现Cannotcreatearecordreaderbecauseofapreviouserror异常:org.apache.spark.SparkException:Jobabortedduetostagefailure:Task0instage0.0failed4times,mostrecentfailure:Losttask0.3instage0.0(TID3,zdwlhadoop1):java.io.IOException:Cannotcreat

spark访问hbase

import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor} import org.apache.hadoop.hbase.mapreduce.TableInputFormat import org.apache.spark.rdd.NewHadoopRDD val conf = HBaseConfiguration.create() conf.set(TableInputFormat.INPUT_TABLE, "tm

spark读取hbase空指针异常,跪求大神指导

问题描述 spark版本:1.2.1hbase版本:0.98importorg.apache.hadoop.hbase.HBaseConfigurationimportorg.apache.hadoop.hbase.mapreduce.TableInputFormatimportorg.apache.spark.SparkConfimportorg.apache.spark.SparkContextobjectHBaseTest{defmain(args:Array[String]){valsp

Hortonworks改进Spark与Hadoop全面整合

http://www.aliyun.com/zixun/aggregation/14112.html">Hortonworks的新代码改进了Spark与Hive的集成,并计划对Spark内存分析平台进行安全和性能方面的升级. Apache Spark内存分析平台如今是大数据分析领域的热门技术,Hadoop发行商Hortonworks近日决定加大对Spark的投入.本周三 Hortonworks宣布其Spark软件将提升Hive集成度,并增加对Spark的ORC数据格式的支持,Hortonw

大数据工具篇之Hive与HBase整合完整教程

一.引言 最近的一次培训,用户特意提到Hadoop环境下HDFS中存储的文件如何才能导入到HBase,关于这部分基于HBase Java API的写入方式,之前曾经有过技术文章共享,本文就不再说明.本文基于Hive执行HDFS批量向HBase导入数据,讲解Hive与HBase的整合问题.这方面的文章已经很多,但是由于版本差异,可操作性不大,本文采用的版本均基于以下版本说明中的版本. 二.版本说明 序号 软件 版本 1 Hive  0.10.0 2 HBase 0.94.0 3 Hadoop 1.

Impala与HBase整合实践

我们知道,HBase是一个基于列的NoSQL数据库,它可以实现的数据的灵活存储.它本身是一个大表,在一些应用中,通过设计RowKey,可以实现对海量数据的快速存储和访问.但是,对于复杂的查询统计类需求,如果直接基于HBase API来实现,性能非常差,或者,可以通过实现MapReduce程序来进行查询分析,这也继承了MapReduce所具备的延迟性. 实现Impala与HBase整合,我们能够获得的好处有如下几个: 可以使用我们熟悉的SQL,像操作传统关系型数据库一样,很容易给出复杂查询.统计分

如何提高spark批量读取HBase数据的性能

问题描述 Configurationconf=HBaseConfiguration.create();StringtableName="testTable";Scanscan=newScan();scan.setCaching(10000);scan.setCacheBlocks(false);conf.set(TableInputFormat.INPUT_TABLE,tableName);ClientProtos.Scanproto=ProtobufUtil.toScan(scan)