这篇文章主要介绍“spark读取hbase的数据实例代码”,在日常操作中,相信很多人在spark读取hbase的数据实例代码问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”spark读取hbase的数据实例代码”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
成都创新互联公司专注于黄山企业网站建设,成都响应式网站建设公司,成都做商城网站。黄山网站建设公司,为黄山等地区提供建站服务。全流程定制网站制作,专业设计,全程项目跟踪,成都创新互联公司专业和态度为您提供的服务package hgs.spark.hbase ///tupian/20230522/52675901 import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.hbase.mapreduce.TableInputFormat import org.apache.hadoop.hbase.client.Scan import org.apache.hadoop.hbase.filter.FilterList import org.apache.hadoop.hbase.filter.FilterList.Operator import org.apache.hadoop.hbase.filter.RowFilter import org.apache.hadoop.hbase.filter.RegexStringComparator import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp import org.apache.hadoop.hbase.protobuf.ProtobufUtil import org.apache.hadoop.hbase.util.Base64 import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.filter.LongComparator object HbaseToSpark { def main(args: Array[String]): Unit = { //System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); val conf = new SparkConf conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") conf.setMaster("local").setAppName("hbasedata") val context = new SparkContext(conf) //hbase配置 val hconf = new HBaseConfiguration hconf.set("hbase.zookeeper.quorum", "bigdata00:2181,bigdata01:2181,bigdata02:2181") hconf.set("hbase.zookeeper.property.clientPort", "2181") hconf.set(TableInputFormat.INPUT_TABLE, "test") val scan = new Scan //扫描的表rowkey的开始和结束 scan.setStartRow("1991".getBytes) scan.setStopRow("3000".getBytes) //val list = new FilterList(Operator.MUST_PASS_ALL) //val filter1 = new RowFilter(CompareOp.GREATER_OR_EQUAL,new LongComparator(1991)) //val filter2 = new RowFilter(CompareOp.LESS_OR_EQUAL,new RegexStringComparator("3000*")) // list.addFilter(filter1) // list.addFilter(filter2) //scan.setFilter(list) //添加scan hconf.set(TableInputFormat.SCAN, convertScanToString(scan)) val hrdd = context.newAPIHadoopRDD(hconf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result]) val resultrdd = hrdd.repartition(2) //打印结果 resultrdd.foreach{case(_,value)=>{ val key = Bytes.toString(value.getRow) val name = Bytes.toString(value.getValue("cf1".getBytes, "name".getBytes)) val age = Bytes.toString(value.getValue("cf1".getBytes, "age".getBytes)) println("rowkey:"+key+" "+"name:"+name+" "+"age:"+age) } } context.stop() } def convertScanToString(scan: Scan) = { val proto = ProtobufUtil.toScan(scan) Base64.encodeBytes(proto.toByteArray) } }
到此,关于“spark读取hbase的数据实例代码”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联-成都网站建设公司网站,小编会继续努力为大家带来更多实用的文章!
售后响应及时
7×24小时客服热线数据备份
更安全、更高效、更稳定价格公道精准
项目经理精准报价不弄虚作假合作无风险
重合同讲信誉,无效全额退款