关键字:SparkSQL读取HBase、SparkSQL自定义外部数据源
前面文章介绍了SparSQL通过Hive操作HBase表。
SparkSQL从1.2开始支持自定义外部数据源(External DataSource),这样就可以通过API接口来实现自己的外部数据源。这里基于Spark1.4.0,简单介绍SparkSQL自定义外部数据源,访问HBase表。
HBase中表如下:
create 'lxw1234',{NAME => 'f1',VERSIONS => 1},{NAME => 'f2',VERSIONS => 1},{NAME => 'f3',VERSIONS => 1} put 'lxw1234','lxw1234.com','f1:c1','name1' put 'lxw1234','lxw1234.com','f1:c2','name2' put 'lxw1234','lxw1234.com','f2:c1','age1' put 'lxw1234','lxw1234.com','f2:c2','age2' put 'lxw1234','lxw1234.com','f3:c1','job1' put 'lxw1234','lxw1234.com','f3:c2','job2' put 'lxw1234','lxw1234.com','f3:c3','job3' hbase(main):025:0* scan 'lxw1234' ROW COLUMN+CELL lxw1234.com column=f1:c1, timestamp=1435624625198, value=name1 lxw1234.com column=f1:c2, timestamp=1435624591717, value=name2 lxw1234.com column=f2:c1, timestamp=1435624608759, value=age1 lxw1234.com column=f2:c2, timestamp=1435624635261, value=age2 lxw1234.com column=f3:c1, timestamp=1435624662282, value=job1 lxw1234.com column=f3:c2, timestamp=1435624697028, value=job2 lxw1234.com column=f3:c3, timestamp=1435624697065, value=job3
进入spark-shell
sh /usr/local/spark-1.4.0-bin-hadoop2.3/bin/spark-shell --jars /tmp/sparksql-hbase.jar --total-executor-cores 30 --executor-memory 4G --master spark://lxw1234.com:7077
运行以下代码:
import sqlContext._ var hbasetable = sqlContext.read.format("com.lxw1234.sparksql.hbase").options(Map( "sparksql_table_schema" -> "(row_key string, c1 string, c2 string, c3 string)", "hbase_table_name" -> "lxw1234", "hbase_table_schema" -> "(:key , f1:c2 , f2:c2 , f3:c3 )" )).load() //sparksql_table_schema参数为sparksql中表的定义 //hbase_table_name参数为HBase中表名 //hbase_table_schema参数为HBase表中需要映射到SparkSQL表中的列族和列,这里映射过//去的字段要和sparksql_table_schema中定义的一致,包括顺序。 scala> hbasetable.printSchema() root |-- row_key: string (nullable = false) |-- c1: string (nullable = false) |-- c2: string (nullable = false) |-- c3: string (nullable = false) hbasetable.registerTempTable("lxw1234") sqlContext.sql("SELECT * from lxw1234").collect res3: Array[org.apache.spark.sql.Row] = Array([lxw1234.com,name2,age2,job3]) sqlContext.sql("SELECT row_key,concat(c1,'|',c2,'|',c3) from lxw1234").collect res3: Array[org.apache.spark.sql.Row] = Array([lxw1234.com,name2|age2|job3])
可以用SQL正常访问。
源码和相关配置
- 本来在SparkSQL中通过外部数据源建表的语法是:
CREATE TEMPORARY TABLE hbasetable
USING com.lxw1234.sparksql.hbase
OPTIONS (
sparksql_table_schema ‘(row_key string, c1 string, c2 string, c3 string)’,
hbase_table_name ‘lxw1234′,
hbase_table_schema ‘(:key , f1:c2 , f2:c2 , f3:c3)’
)
在我的Spark1.4中报错,会使用Hive的语法解析器解析这个DDL语句,因为Hive0.13中没有这种语法,因此报错。
是否是因为Spark1.4包的编译了Hive的原因?
- 基于Spark1.4的程序源码com.lxw1234.sparksql.hbase见:http://lxw1234.com/archives/2015/07/332.htm
- 源码的编译依赖HBase的相关jar包:
hbase-client-0.96.1.1-cdh5.0.0.jar
hbase-common-0.96.1.1-cdh5.0.0.jar
hbase-protocol-0.96.1.1-cdh5.0.0.jar
hbase-server-0.96.1.1-cdh5.0.0.jar
还有HBase的集群信息:
hbase.zookeeper.quorum
hbase.client.scanner.caching
我之前在配置时候已经将这几个jar包和参数加到Spark集群的CLASSPATH中了,可参考 http://lxw1234.com/archives/2015/07/330.htm
- 此程序是OopsOutOfMemory基于Spark1.2开发的,我只做了很小的修改。
https://github.com/OopsOutOfMemory/spark-sql-hbase
- 此程序只做学习和测试使用,并未测试性能。
相关推荐
使用spark读取hbase中的数据,并插入到mysql中
创建`JavaSparkContext`后,我们可以使用`sc.newAPIHadoopRDD()`方法创建一个RDD(弹性分布式数据集),这个RDD将直接读取HBase中的数据。传入的参数包括配置、`TableInputFormat`的类名以及键值对的类型。在这个...
本主题将详细探讨如何利用Java通过Thrift-0.9.1版本来读取HBase表数据。 HBase是一个基于Google Bigtable设计的开源NoSQL数据库,它构建在Hadoop之上,提供高可靠性、高性能、分布式的行存储。HBase支持实时读写,...
hbasesink 自定义序列化类 ,可实现自定义rowkey及去除字段两边索引,具体请看下代码。 hbasesink 自定义序列化类 ,可实现自定义rowkey及去除字段两边索引,具体请看下代码。
通过结合上述两段代码,你可以实现从HBase中读取数据并写入HDFS的功能。这只是一个基本的实现,实际应用中可能需要处理更复杂的情况,例如批量读取、错误处理、数据转换等。同时,为了提高性能,你还可以考虑使用...
因工作需要用python通过hbase的thrift服务读取Hbase表数据,发现公司的测试环境还不支持,于是自己动手准备环境,在此我将在安装步骤尽可能描述清楚,旨在给第一次动手安装的朋友,此过程亲测成功! 安装过程如下: ...
Kettle支持多种数据源,并且可以通过图形界面设计数据流,非常适合大规模数据处理场景。 ##### 1.1 集群架构 Kettle集群主要由一个主Carte服务器和多个从Carte服务器组成,其结构类似于master-slave模式。主Carte...
3. 查询功能实现:根据RowKey查询数据是HBase的基本操作,通过输入RowKey,后台执行get操作获取对应行数据,并展示在页面上。 4. 表管理:支持HBase的建表和删除操作,这需要调用HBase的Admin API,完成表的创建、...
HBASE的一个读取数据流程的解析,清晰的画出整个过程,十分有利于理解
- 分区键(Partition Key):通过配置HBase的分区键,可以实现数据的预分区,有助于提高数据分布的均匀性,从而提高查询性能。 - 压缩:HBase支持多种数据压缩算法(如Snappy、GZIP等),能够显著减少存储空间,提高...
在Java编程环境中,HBase是一种分布式、高性能的NoSQL数据库,常用于大数据处理。本示例代码主要展示了如何使用Java API连接HBase数据库,并执行基本的CRUD(创建、读取、更新、删除)操作,同时也包括了批量操作的...
1. Hive到HBase:利用Hive的外部表功能,将Hive表的数据导出到HBase。这通常通过定义一个HBase的Hive表,并使用HBaseStorageHandler来实现。 2. HBase到Hive:通过MapReduce任务,将HBase中的数据导入到Hive,创建...
- 编写代码,使用Table和Put对象将数据从本地文件读取并写入到HBase表中。 - 编译并运行Java程序,完成数据导入。 在整个过程中,确保所有组件的版本兼容,例如HBase与Hadoop、Sqoop与Hadoop之间的版本匹配。同时...
tsv格式的数据库测试文件,hbase可以通过采用importtsv导入外部数据到hbase中
Spark作为一个快速、通用的大数据处理引擎,可以高效地读取和写入HBase这种分布式NoSQL数据库。下面将详细介绍如何使用Spark及pyspark进行HBase数据的读写,以及HBase的安装配置。 首先,要进行HBase的安装与配置。...