spark访问hbase
发表于:2026-10-31 作者:千家信息网编辑
千家信息网最后更新 2026年10月31日,import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}import org.apache.hadoop.hbase.
千家信息网最后更新 2026年10月31日spark访问hbase
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}import org.apache.hadoop.hbase.mapreduce.TableInputFormatimport org.apache.spark.rdd.NewHadoopRDDval conf = HBaseConfiguration.create()conf.set(TableInputFormat.INPUT_TABLE, "tmp")var hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])hBaseRDD.count()import scala.collection.JavaConverters._hBaseRDD.map(tuple => tuple._2).map(result => result.getColumn("cf".getBytes(), "val".getBytes())).map(keyValues => {( keyValues.asScala.reduceLeft { (a, b) => if (a.getTimestamp > b.getTimestamp) a else b }.getRow, keyValues.asScala.reduceLeft { (a, b) => if (a.getTimestamp > b.getTimestamp) a else b }.getValue)}).take(10)hBaseRDD.map(tuple => tuple._2).map(result => (result.getRow, result.getColumn("cf".getBytes(), "val".getBytes()))).map(row => {( row._1.map(_.toChar).mkString, row._2.asScala.reduceLeft { (a, b) => if (a.getTimestamp > b.getTimestamp) a else b }.getValue.map(_.toChar).mkString)}).take(10)conf.set(TableInputFormat.INPUT_TABLE, "test1")//var hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])hBaseRDD.map(tuple => tuple._2).map(result => (result.getRow, result.getColumn("lf".getBytes(), "app1".getBytes()))).map(row => if (row._2.size > 0) {( row._1.map(_.toChar).mkString, row._2.asScala.reduceLeft { (a, b) => if (a.getTimestamp > b.getTimestamp) a else b }.getValue.map(_.toInt).mkString)}).take(10)import java.nio.ByteBufferhBaseRDD.map(tuple => tuple._2).map(result => (result.getRow, result.getColumn("lf".getBytes(), "app1".getBytes()))).map(row => if (row._2.size > 0) {( row._1.map(_.toChar).mkString, ByteBuffer.wrap(row._2.asScala.reduceLeft { (a, b) => if (a.getTimestamp > b.getTimestamp) a else b }.getValue).getLong)}).take(10)//conf.set(TableInputFormat.SCAN_COLUMN_FAMILY, "lf")conf.set(TableInputFormat.SCAN_COLUMNS, "lf:app1")//var hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])import java.nio.ByteBufferhBaseRDD.map(tuple => tuple._2).map(result => { ( result.getRow.map(_.toChar).mkString, ByteBuffer.wrap(result.value).getLong )}).take(10)val conf = HBaseConfiguration.create()conf.set(TableInputFormat.INPUT_TABLE, "test1")var hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])var rows = hBaseRDD.map(tuple => tuple._2).map(result => result.getRow.map(_.toChar).mkString)rows.map(row => row.split("\\|")).map(r => if (r.length > 1) (r(0), r(1)) else (r(0), "") ).groupByKey.take(10)
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全
广州淘业互联网络科技
朗新科技与工业互联网
深圳互联网科技公司
成都互联网科技公司
鑫亿盛互联网科技
撒么互联网科技公司
人人互联网科技是培训
互联网科技股有哪些
济南和联互联网科技有限公司
阿里互联网科技公司
互联网公司科技股
千方科技属于互联网企业吗
华夏恒生互联网科技013171
互联网数据科技专业
卡尔迅互联网科技有限公司
山东大美朱楼互联网科技有限公司
北京军民互联网科技公司
绵阳真晨互联网科技有限公司
保山浩联互联网科技有限公司
六子台科技互联网展会
咖啡
云南咖啡
中国互联网科技股
智网互联网科技
齐家互联网科技有限公司
哈尔滨新华互联网科技
广州新华互联网科技学校电竞方面
领航科技大厦互联网公司
宁波跨境商互联网科技有限公司
互联网科技平台产品运营