Spark与HBase的整合

前言

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

虽然我比较推崇实时计算的,然而补数据或者需要计算历史数据的时候,批处理就少不了。这里有两个选择,一个是基于HBase的行为数据进行计算,或者基于原始的Hive数据进行计算,最终选择了前者,这里就涉及到Spark(StreamingPro) 对HBase的操作了。

难点

和Spark 整合,意味着最好能有Schema(Mapping),但是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函数之类的做处理了。 这是第二种方案的具体做法。这个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
  }
}

而如果要实现第一种方案,可以将数据转化为JSON数据源,这样就可以利用Spark JSON自动推倒Schema的能力了,当然,这种方式对性能上会影响比较大,因为Schema推测其实便利了一边数据的。

稿源:祝威廉 (源链) | 关于 | 阅读提示

本站遵循[CC BY-NC-SA 4.0]。如您有版权、意见投诉等问题,请通过eMail联系我们处理。
酷辣虫 » 后端存储 » Spark与HBase的整合

喜欢 (0)or分享给?

专业 x 专注 x 聚合 x 分享 CC BY-NC-SA 4.0

使用声明 | 英豪名录

登录

忘记密码 ?

切换登录

注册