spark将数据写入hbase以及从hbase读取数据

2024-08-27 10:32

本文主要是介绍spark将数据写入hbase以及从hbase读取数据,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

原文:http://blog.csdn.net/u013468917/article/details/52822074

本文将介绍

1、spark如何利用saveAsHadoopDataset和saveAsNewAPIHadoopDataset将RDD写入hbase

2、spark从hbase中读取数据并转化为RDD

操作方式为在eclipse本地运行spark连接到远程的hbase。

java版本:1.7.0

scala版本:2.10.4

zookeeper版本:3.4.5(禁用了hbase自带zookeeper,选择自己部署的)

hadoop版本:2.4.1

spark版本:1.6.1

hbase版本:1.2.3

集群:centos6.5_x64

将RDD写入hbase

注意点:

依赖:

将lib目录下的hadoop开头jar包、hbase开头jar包添加至classpath

此外还有lib目录下的:zookeeper-3.4.6.jar、metrics-core-2.2.0.jar(缺少会提示hbase RpcRetryingCaller: Call exception不断尝试重连hbase,不报错)、htrace-core-3.1.0-incubating.jar、guava-12.0.1.jar

$SPARK_HOME/lib目录下的 spark-assembly-1.6.1-hadoop2.4.0.jar

不同的package中可能会有相同名称的类,不要导错

连接集群:

spark应用需要连接到zookeeper集群,然后借助zookeeper访问hbase。一般可以通过两种方式连接到zookeeper:

第一种是将hbase-site.xml文件加入classpath

第二种是在HBaseConfiguration实例中设置

如果不设置,默认连接的是localhost:2181会报错:connection refused 

本文使用的是第二种方式。

hbase创建表:

虽然可以在spark应用中创建hbase表,但是不建议这样做,最好在hbase shell中创建表,spark写或读数据

使用saveAsHadoopDataset写入数据

[plain] view plain copy
  1. package com.test  
  2.   
  3. import org.apache.hadoop.hbase.HBaseConfiguration  
  4. import org.apache.hadoop.hbase.client.Put  
  5. import org.apache.hadoop.hbase.io.ImmutableBytesWritable  
  6. import org.apache.hadoop.hbase.mapred.TableOutputFormat  
  7. import org.apache.hadoop.hbase.util.Bytes  
  8. import org.apache.hadoop.mapred.JobConf  
  9. import org.apache.spark.SparkConf  
  10. import org.apache.spark.SparkContext  
  11. import org.apache.spark.rdd.RDD.rddToPairRDDFunctions  
  12.   
  13. object TestHBase {  
  14.   
  15.   def main(args: Array[String]): Unit = {  
  16.     val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")  
  17.     val sc = new SparkContext(sparkConf)  
  18.   
  19.     val conf = HBaseConfiguration.create()  
  20.     //设置zooKeeper集群地址,也可以通过将hbase-site.xml导入classpath,但是建议在程序里这样设置  
  21.     conf.set("hbase.zookeeper.quorum","slave1,slave2,slave3")  
  22.     //设置zookeeper连接端口,默认2181  
  23.     conf.set("hbase.zookeeper.property.clientPort", "2181")  
  24.   
  25.     val tablename = "account"  
  26.       
  27.     //初始化jobconf,TableOutputFormat必须是org.apache.hadoop.hbase.mapred包下的!  
  28.     val jobConf = new JobConf(conf)  
  29.     jobConf.setOutputFormat(classOf[TableOutputFormat])  
  30.     jobConf.set(TableOutputFormat.OUTPUT_TABLE, tablename)  
  31.       
  32.     val indataRDD = sc.makeRDD(Array("1,jack,15","2,Lily,16","3,mike,16"))  
  33.   
  34.   
  35.     val rdd = indataRDD.map(_.split(',')).map{arr=>{  
  36.       /*一个Put对象就是一行记录,在构造方法中指定主键  
  37.        * 所有插入的数据必须用org.apache.hadoop.hbase.util.Bytes.toBytes方法转换  
  38.        * Put.add方法接收三个参数:列族,列名,数据  
  39.        */  
  40.       val put = new Put(Bytes.toBytes(arr(0).toInt))  
  41.       put.add(Bytes.toBytes("cf"),Bytes.toBytes("name"),Bytes.toBytes(arr(1)))  
  42.       put.add(Bytes.toBytes("cf"),Bytes.toBytes("age"),Bytes.toBytes(arr(2).toInt))  
  43.       //转化成RDD[(ImmutableBytesWritable,Put)]类型才能调用saveAsHadoopDataset  
  44.       (new ImmutableBytesWritable, put)   
  45.     }}  
  46.       
  47.     rdd.saveAsHadoopDataset(jobConf)  
  48.       
  49.     sc.stop()  
  50.   }  
  51.   
  52. }  

使用saveAsNewAPIHadoopDataset写入数据


[plain] view plain copy
  1. package com.test  
  2.   
  3. import org.apache.hadoop.hbase.HBaseConfiguration  
  4. import org.apache.hadoop.hbase.mapreduce.TableOutputFormat  
  5. import org.apache.spark._  
  6. import org.apache.hadoop.mapreduce.Job  
  7. import org.apache.hadoop.hbase.io.ImmutableBytesWritable  
  8. import org.apache.hadoop.hbase.client.Result  
  9. import org.apache.hadoop.hbase.client.Put  
  10. import org.apache.hadoop.hbase.util.Bytes  
  11.   
  12. object TestHBase3 {  
  13.   
  14.   def main(args: Array[String]): Unit = {  
  15.     val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")  
  16.     val sc = new SparkContext(sparkConf)  
  17.       
  18.     val tablename = "account"  
  19.       
  20.     sc.hadoopConfiguration.set("hbase.zookeeper.quorum","slave1,slave2,slave3")  
  21.     sc.hadoopConfiguration.set("hbase.zookeeper.property.clientPort", "2181")  
  22.     sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE, tablename)  
  23.       
  24.     val job = new Job(sc.hadoopConfiguration)  
  25.     job.setOutputKeyClass(classOf[ImmutableBytesWritable])  
  26.     job.setOutputValueClass(classOf[Result])    
  27.     job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])    
  28.   
  29.     val indataRDD = sc.makeRDD(Array("1,jack,15","2,Lily,16","3,mike,16"))  
  30.     val rdd = indataRDD.map(_.split(',')).map{arr=>{  
  31.       val put = new Put(Bytes.toBytes(arr(0)))  
  32.       put.add(Bytes.toBytes("cf"),Bytes.toBytes("name"),Bytes.toBytes(arr(1)))  
  33.       put.add(Bytes.toBytes("cf"),Bytes.toBytes("age"),Bytes.toBytes(arr(2).toInt))  
  34.       (new ImmutableBytesWritable, put)   
  35.     }}  
  36.       
  37.     rdd.saveAsNewAPIHadoopDataset(job.getConfiguration())  
  38.   }  
  39.   
  40. }  


从hbase读取数据转化成RDD

本例基于官方提供的例子

[plain] view plain copy
  1. package com.test  
  2.   
  3. import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}  
  4. import org.apache.hadoop.hbase.client.HBaseAdmin  
  5. import org.apache.hadoop.hbase.mapreduce.TableInputFormat  
  6. import org.apache.spark._  
  7. import org.apache.hadoop.hbase.client.HTable  
  8. import org.apache.hadoop.hbase.client.Put  
  9. import org.apache.hadoop.hbase.util.Bytes  
  10. import org.apache.hadoop.hbase.io.ImmutableBytesWritable  
  11. import org.apache.hadoop.hbase.mapreduce.TableOutputFormat  
  12. import org.apache.hadoop.mapred.JobConf  
  13. import org.apache.hadoop.io._  
  14.   
  15. object TestHBase2 {  
  16.   
  17.   def main(args: Array[String]): Unit = {  
  18.     val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local")  
  19.     val sc = new SparkContext(sparkConf)  
  20.       
  21.     val tablename = "account"  
  22.     val conf = HBaseConfiguration.create()  
  23.     //设置zooKeeper集群地址,也可以通过将hbase-site.xml导入classpath,但是建议在程序里这样设置  
  24.     conf.set("hbase.zookeeper.quorum","slave1,slave2,slave3")  
  25.     //设置zookeeper连接端口,默认2181  
  26.     conf.set("hbase.zookeeper.property.clientPort", "2181")  
  27.     conf.set(TableInputFormat.INPUT_TABLE, tablename)  
  28.   
  29.     // 如果表不存在则创建表  
  30.     val admin = new HBaseAdmin(conf)  
  31.     if (!admin.isTableAvailable(tablename)) {  
  32.       val tableDesc = new HTableDescriptor(TableName.valueOf(tablename))  
  33.       admin.createTable(tableDesc)  
  34.     }  
  35.   
  36.     //读取数据并转化成rdd  
  37.     val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],  
  38.       classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],  
  39.       classOf[org.apache.hadoop.hbase.client.Result])  
  40.   
  41.     val count = hBaseRDD.count()  
  42.     println(count)  
  43.     hBaseRDD.foreach{case (_,result) =>{  
  44.       //获取行键  
  45.       val key = Bytes.toString(result.getRow)  
  46.       //通过列族和列名获取列  
  47.       val name = Bytes.toString(result.getValue("cf".getBytes,"name".getBytes))  
  48.       val age = Bytes.toInt(result.getValue("cf".getBytes,"age".getBytes))  
  49.       println("Row key:"+key+" Name:"+name+" Age:"+age)  
  50.     }}  
  51.   
  52.     sc.stop()  
  53.     admin.close()  
  54.   }  
  55. }  


这篇关于spark将数据写入hbase以及从hbase读取数据的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/1111416

相关文章

MySQL中查询和展示LONGBLOB类型数据的技巧总结

《MySQL中查询和展示LONGBLOB类型数据的技巧总结》在MySQL中LONGBLOB是一种二进制大对象(BLOB)数据类型,用于存储大量的二进制数据,:本文主要介绍MySQL中查询和展示LO... 目录前言1. 查询 LONGBLOB 数据的大小2. 查询并展示 LONGBLOB 数据2.1 转换为十

使用SpringBoot+InfluxDB实现高效数据存储与查询

《使用SpringBoot+InfluxDB实现高效数据存储与查询》InfluxDB是一个开源的时间序列数据库,特别适合处理带有时间戳的监控数据、指标数据等,下面详细介绍如何在SpringBoot项目... 目录1、项目介绍2、 InfluxDB 介绍3、Spring Boot 配置 InfluxDB4、I

Java整合Protocol Buffers实现高效数据序列化实践

《Java整合ProtocolBuffers实现高效数据序列化实践》ProtocolBuffers是Google开发的一种语言中立、平台中立、可扩展的结构化数据序列化机制,类似于XML但更小、更快... 目录一、Protocol Buffers简介1.1 什么是Protocol Buffers1.2 Pro

Python实现数据可视化图表生成(适合新手入门)

《Python实现数据可视化图表生成(适合新手入门)》在数据科学和数据分析的新时代,高效、直观的数据可视化工具显得尤为重要,下面:本文主要介绍Python实现数据可视化图表生成的相关资料,文中通过... 目录前言为什么需要数据可视化准备工作基本图表绘制折线图柱状图散点图使用Seaborn创建高级图表箱线图热

MySQL数据脱敏的实现方法

《MySQL数据脱敏的实现方法》本文主要介绍了MySQL数据脱敏的实现方法,包括字符替换、加密等方法,通过工具类和数据库服务整合,确保敏感信息在查询结果中被掩码处理,感兴趣的可以了解一下... 目录一. 数据脱敏的方法二. 字符替换脱敏1. 创建数据脱敏工具类三. 整合到数据库操作1. 创建服务类进行数据库

MySQL中处理数据的并发一致性的实现示例

《MySQL中处理数据的并发一致性的实现示例》在MySQL中处理数据的并发一致性是确保多个用户或应用程序同时访问和修改数据库时,不会导致数据冲突、数据丢失或数据不一致,MySQL通过事务和锁机制来管理... 目录一、事务(Transactions)1. 事务控制语句二、锁(Locks)1. 锁类型2. 锁粒

Qt中实现多线程导出数据功能的四种方式小结

《Qt中实现多线程导出数据功能的四种方式小结》在以往的项目开发中,在很多地方用到了多线程,本文将记录下在Qt开发中用到的多线程技术实现方法,以导出指定范围的数字到txt文件为例,展示多线程不同的实现方... 目录前言导出文件的示例工具类QThreadQObject的moveToThread方法实现多线程QC

SpringBoot集成EasyExcel实现百万级别的数据导入导出实践指南

《SpringBoot集成EasyExcel实现百万级别的数据导入导出实践指南》本文将基于开源项目springboot-easyexcel-batch进行解析与扩展,手把手教大家如何在SpringBo... 目录项目结构概览核心依赖百万级导出实战场景核心代码效果百万级导入实战场景监听器和Service(核心

使用Python开发一个Ditto剪贴板数据导出工具

《使用Python开发一个Ditto剪贴板数据导出工具》在日常工作中,我们经常需要处理大量的剪贴板数据,下面将介绍如何使用Python的wxPython库开发一个图形化工具,实现从Ditto数据库中读... 目录前言运行结果项目需求分析技术选型核心功能实现1. Ditto数据库结构分析2. 数据库自动定位3

pandas数据的合并concat()和merge()方式

《pandas数据的合并concat()和merge()方式》Pandas中concat沿轴合并数据框(行或列),merge基于键连接(内/外/左/右),concat用于纵向或横向拼接,merge用于... 目录concat() 轴向连接合并(1) join='outer',axis=0(2)join='o