Spark On HBase

栏目: 数据库 · 发布时间: 6年前

内容简介:MapReduce早已经对接了HBase,以HBase作为数据源,完成批量数据的读写。如今继MapReduce之后的Spark在大数据领域有着举足轻重的地位,无论跑批,流处理,甚至图计算等都有它的用武之地。Spark对接HBase成为不少用户的需求。Spark和HBase无缝对接意味着我们不再需要关心安全和RDD与HBase交互的细节。更方便应用Spark带来的批处理,流处理等能力。比如以下常见的应用场景:目前已经有多种Spark对接HBase的实现,这里我们选取三个有代表的工作进行分析:

一.前言

MapReduce早已经对接了HBase,以HBase作为数据源,完成批量数据的读写。如今继MapReduce之后的Spark在大数据领域有着举足轻重的地位,无论跑批,流处理,甚至图计算等都有它的用武之地。Spark对接HBase成为不少用户的需求。

二.Spark On HBase

1.可以解决的问题

Spark和HBase无缝对接意味着我们不再需要关心安全和RDD与HBase交互的细节。更方便应用Spark带来的批处理,流处理等能力。比如以下常见的应用场景:

  1. 以HBase作为存储,通过Spark对流式数据处理。

  2. 以HBase作为存储,完成大规模的图或者DAG的计算。

  3. 通过Spark对HBase做BulkLoad操作

  4. 同Spark SQL对HBase数据做交互式分析

2.社区相关的工作

目前已经有多种Spark对接HBase的实现,这里我们选取三个有代表的工作进行分析:

2.1 华为: Spark-SQL-on-HBase

特点:
扩展了Spark SQL的parse功能来对接HBase。通过coprocessor和自定义filter来提升读写性能。

优点:

  • 扩展了对应的cli功能,支持Scala shell和Python shell

  • 多种性能优化方式,甚至支持sub plan到coprocessor实现partial aggregation.

  • 支持 Java 和Python API

  • 支持r ow key 组合

  • 支持常用DDL和DML(包括bulkload,但不支持update)

缺点:

  • 不支持支持基于时间戳和版本的查询

  • 不支持安全

  • row key支持原始类型或者String,不支持复杂数据类型

使用示例:

在HBase中创建表,并写入数据

$HBase_Home/bin/hbase shell
create 'hbase_numbers', 'f'for i in '1'..'100' do for j in '1'..'2' do put 'hbase_numbers', "row#{i}", "f:c#{j}", "#{i}#{j}" end end

使用 Spark SQL 创建表并与HBase表建立映射

$SPARK_HBASE_Home/bin/hbase-sqlCREATE TABLE numbers
rowkey STRING, a STRING, b STRING, PRIMARY KEY (rowkey)
MAPPED BY hbase_numbers COLS=[a=f.c1, b=f.c2];

查询

select a, b from numbers where b > "980"

2.2 Hortonworks: Apache HBase Connector

特点:
以简单的方式实现了标准的Spark Datasource API,使用Spark Catalyst引擎做查询优化。同时通过scratch来构建RDD,也实现了许多常见的查询优化。

优点:

  • native avro支持

  • 谓词下推和分区裁剪

  • 支持row key组合

  • 支持安全

缺点:

  • SQL语法不够丰富,只支持spark sql原有的语法

  • 只支持java原始类型

  • 不支持多语言API

使用示例:

定义 HBase Catalog

def catalog = s"""{
        |"table":{"namespace":"default", "name":"table1"},
        |"rowkey":"key",
        |"columns":{
          |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
          |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
          |"col2":{"cf":"cf2", "col":"col2", "type":"double"},
          |"col3":{"cf":"cf3", "col":"col3", "type":"float"},
          |"col4":{"cf":"cf4", "col":"col4", "type":"int"},
          |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
          |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
          |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
          |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
        |}
      |}""".stripMargin

使用 SQL 查询

// Load the dataframeval df = withCatalog(catalog)//SQL exampledf.createOrReplaceTempView("table")
sqlContext.sql("select count(col1) from table").show

2.3 Cloudrea: SparkOnHBase

特点:
通过简单的接口实现链接Spark与HBase, 支持常用的bulk读写。架构图如下:

Spark On HBase

优点

  • 支持安全

  • 通过get或者scan直接生成rdd, 并可以使用API完成更高级的功能

  • 支持组合rowkey

  • 支持多种bulk操作

  • 为spark和 spark streaming提供相似的API

  • 支持谓词下推优化

缺点

  • 不支持复杂数据类型

  • SQL只支持spark sql原有的语法

使用示例

直接使用scan创建一个RDD

SparkConf sparkConf = new SparkConf().setAppName(  
                "Scan_RDD").set("spark.executor.memory", "2000m").setMaster(  
                "spark://xx.xx.xx.xx:7077")
                 .setJars(new String[]{"/path/to/hbase.jar"});  

val sc = new SparkContext(sparkConf)

val conf = HBaseConfiguration.create()

val hbaseContext = new HBaseContext(sc, conf)var scan = new Scan()
scan.setCaching(100)var getRdd = hbaseContext.hbaseRDD(tableName, scan)

创建一个RDD并把RDD的内容写入HBase

val sc = new SparkContext(sparkConf)//This is making a RDD of//(RowKey, columnFamily, columnQualifier, value)val rdd = sc.parallelize(Array(
      (Bytes.toBytes("1"), Array((Bytes.toBytes(columnFamily), Bytes.toBytes("1"), Bytes.toBytes("1")))),
      (Bytes.toBytes("2"), Array((Bytes.toBytes(columnFamily), Bytes.toBytes("1"), Bytes.toBytes("2")))),
      (Bytes.toBytes("3"), Array((Bytes.toBytes(columnFamily), Bytes.toBytes("1"), Bytes.toBytes("3")))),
      (Bytes.toBytes("4"), Array((Bytes.toBytes(columnFamily), Bytes.toBytes("1"), Bytes.toBytes("4")))),
      (Bytes.toBytes("5"), Array((Bytes.toBytes(columnFamily), Bytes.toBytes("1"), Bytes.toBytes("5"))))
     )
    )//Create the HBase config like you normally would  then//Pass the HBase configs and SparkContext to the HBaseContextval conf = HBaseConfiguration.create();
val hbaseContext = new HBaseContext(sc, conf);//Now give the rdd, table name, and a function that will convert a RDD record to a put, and finally// A flag if you want the puts to be batchedhbaseContext.bulkPut[(Array[Byte], Array[(Array[Byte], Array[Byte], Array[Byte])])](rdd,
    tableName,    //This function is really important because it allows our source RDD to have data of any type
    // Also because puts are not serializable
    (putRecord) > {
      val put = new Put(putRecord._1)
      putRecord._2.foreach((putValue) > put.add(putValue._1, putValue._2, putValue._3))
       put
    },    true);

2.4 综合对比

产品 SQL支持优化 支持安全 接口丰富易用度 易集成到HBase 社区活跃度
华为 近两年无更新
Hortonworks 较多 近一个月内有更新
Cloudrea 较高 已集成到HBASE trunk且持续更新

3. 最后

社区中有不少Spark on HBase的工作,出发点都是为了提供更易用,更高效的接口。其中Cloudrea的SparkOnHbase更加灵活简单,在2015年8月被提交到HBase的主干(trunk)上,模块名为HBase-Spark Module,目前准备在HBASE 2.0 正式Release, 相信这个特性一定是HBase新版本的一个亮点。 于此同时云HBase也会与社区同步发展,使用包括但不限于Spark On HBase的新特性,届时欢迎大家尝鲜。如若文章中有不准确的描述,请多多指正,谢谢!

4. 参考

https://hortonworks.com/blog/spark-hbase-dataframe-based-hbase-connector/
http://blog.cloudera.com/blog/2014/12/new-in-cloudera-labs-sparkonhbase/
https://issues.apache.org/jira/browse/HBASE-13992
http://blog.madhukaraphatak.com/introduction-to-spark-two-part-6/
https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-sql-catalyst.htmlh

大家工作学习遇到HBase技术问题,把问题发布到HBase技术社区论坛 http://hbase.group ,欢迎大家论坛上面提问留言讨论。想了解更多HBase技术关注HBase技术社区公众号(微信号:hbasegroup),非常欢迎大家积极投稿。

Spark On HBase

长按下面的二维码加入HBase技术社区微信群

Spark On HBase


以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

谷歌的断舍离:互联网企业的破坏式创新

谷歌的断舍离:互联网企业的破坏式创新

[日]辻野晃一郎 / 樊颖 / 机械工业出版社 / 2018-1 / 45.00

本书主要分为三部分: 第一部分主要讨论了世界当下如火如荼的互联网企业进军传统产业大潮,并探讨了传统企业在互联网时代的救赎之路。 第二部分主要探讨了成功体验的反面:速度与迭代,并讨论了传统企业之所以无法实现迭代与快速发展的关键原因。介绍互联网公司如何通过组织精简流程来实现高速竞争时代的机动性。 第三部分讨论了互联网时代究竟需要什么样的人才,传统企业的员工应当怎样投身互联网企业才能避......一起来看看 《谷歌的断舍离:互联网企业的破坏式创新》 这本书的介绍吧!

JS 压缩/解压工具
JS 压缩/解压工具

在线压缩/解压 JS 代码

随机密码生成器
随机密码生成器

多种字符组合密码

Base64 编码/解码
Base64 编码/解码

Base64 编码/解码