文档章节

Spark 批量存取 HBase

 大胖和二胖
发布于 2016/08/26 14:20
字数 374
阅读 462
收藏 0

FileAna.scala

object FileAna {

  //  val conf: Configuration = HBaseConfiguration.create()

  val hdfsPath = "hdfs://master:9000"
  val hdfs = FileSystem.get(new URI(hdfsPath), new Configuration())

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("FileAna").setMaster("spark://master:7077").
      set("spark.driver.host", "192.168.1.127").
      setJars(List("/home/pang/woozoomws/spark-service.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-common-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-client-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-protocol-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/htrace-core-3.1.0-incubating.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-server-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/metrics-core-2.2.0.jar"))
    val sc = new SparkContext(conf)
    val rdd = sc.textFile("hdfs://master:9000/woozoom/msgfile.txt")
    val rdd2 = rdd.map(x => convertToHbase(anaMavlink(x)))

    val hbaseConf = HBaseConfiguration.create()
    hbaseConf.addResource("/home/hadoop/software/hbase-1.2.2/conf/hbase-site.xml");

    val jobConf = new JobConf(hbaseConf, this.getClass)
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, "MissionItem")

    rdd2.saveAsHadoopDataset(jobConf)

    sc.stop()
  }

  def convertScanToString(scan: Scan) = {
    val proto = ProtobufUtil.toScan(scan)
    Base64.encodeBytes(proto.toByteArray)
  }

  def convertToHbase(msg: MAVLinkMessage) = {
    val p = new Put(Bytes.toBytes(UUID.randomUUID().toString()))
    if (msg.isInstanceOf[msg_mission_item]) {
      val missionItem = msg.asInstanceOf[msg_mission_item]
      p.addColumn(Bytes.toBytes("data"), Bytes.toBytes("x"), Bytes.toBytes(missionItem.x))
      p.addColumn(Bytes.toBytes("data"), Bytes.toBytes("y"), Bytes.toBytes(missionItem.y))
      p.addColumn(Bytes.toBytes("data"), Bytes.toBytes("z"), Bytes.toBytes(missionItem.z))
    }
    (new ImmutableBytesWritable, p)
  }

  val anaMavlink = (str: String) => {
    val bytes = ByteAndHex.hexStringToBytes(str)
    QuickParser.parse(bytes).unpack()
  }
}

ReadHBase.scala

object ReadHBase {

  //  val conf: Configuration = HBaseConfiguration.create()

  val hdfsPath = "hdfs://master:9000"
  val hdfs = FileSystem.get(new URI(hdfsPath), new Configuration())

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("FileAna").setMaster("spark://master:7077").
      set("spark.driver.host", "192.168.1.127").
      setJars(List("/home/pang/woozoomws/spark-service.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-common-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-client-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-protocol-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/htrace-core-3.1.0-incubating.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/hbase-server-1.2.2.jar",
        "/home/pang/woozoomws/spark-service/lib/hbase/metrics-core-2.2.0.jar"))
    val sc = new SparkContext(conf)

    val hbaseConf = HBaseConfiguration.create()
    hbaseConf.addResource("/home/hadoop/software/hbase-1.2.2/conf/hbase-site.xml");

    hbaseConf.set(TableInputFormat.INPUT_TABLE, "MissionItem")
    val scan = new Scan()
    hbaseConf.set(TableInputFormat.SCAN, convertScanToString(scan))
    val readRDD = sc.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])

    val count = readRDD.count()
    println("Mission Item Count:" + count)

    sc.stop()
  }

  def convertScanToString(scan: Scan) = {
    val proto = ProtobufUtil.toScan(scan)
    Base64.encodeBytes(proto.toByteArray)
  }
}

© 著作权归作者所有

粉丝 23
博文 69
码字总数 50842
作品 0
沈阳
架构师
私信 提问
HBase实操 | 使用Spark通过BulkLoad快速导入数据到HBase

HBase社区直播本期分享专家:明惠(网名:过往记忆)-阿里云数据架构师 视频地址: https://yq.aliyun.com/live/590?spm=a2c4e.11155435.0.0.460177969kCLxf PPT地址: https://yq.aliyun.com/...

hbase小能手
2018/11/16
0
0
Spark Streaming入门

欢迎大家前往腾讯云+社区,获取更多腾讯海量技术实践干货哦~ 本文将帮助您使用基于HBase的Apache Spark Streaming。Spark Streaming是Spark API核心的一个扩展,支持连续的数据流处理。 什么...

腾讯云加社区
2018/05/16
0
0
HBase技术双周刊第1期:HBase 在人工智能场景的使用

精彩直播 HBase多模式 HBase多模式,包括 分析层:支持复杂分析、算子下推;多模式层:提供各种模型转换,贴切业务;索引引擎:提供索引支持,基于 Lucene ;存储引擎:提供 KV 支持,基于L...

萌萌怪兽
2018/11/22
0
0
云栖大会 | Apache Spark 3.0 和 Koalas 最新进展

本资料来自2019-09-26在杭州举办的云栖大会的大数据 & AI 峰会分会。议题名称《New Developments in the Open Source Ecosystem: Apache Spark 3.0 and Koalas》,分享嘉宾李潇,Databricks ...

Delta Lake
09/27
0
0
HBase+Spark技术双周刊第1期

精彩直播 HBase多模式 HBase多模式,包括 分析层:支持复杂分析、算子下推;多模式层:提供各种模型转换,贴切业务;索引引擎:提供索引支持,基于 Lucene ;存储引擎:提供 KV 支持,基于L...

hbase小能手
2018/11/23
0
0

没有更多内容

加载失败,请刷新页面

加载更多

学习记录(day05-标签操作、属性绑定、语句控制、数据绑定、事件绑定、案例用户登录)

[TOC] 1.1.1标签操作v-text&v-html v-text:会把data中绑定的数据值原样输出。 v-html:会把data中值输出,且会自动解析html代码 <!--可以将指定的内容显示到标签体中--><标签 v-text=""></......

庭前云落
42分钟前
5
0
VMware vSphere的两种RDM磁盘

在VMware vSphere vCenter中创建虚拟机时,可以添加一种叫RDM的磁盘。 RDM - Raw Device Mapping,原始设备映射,那么,RDM磁盘是不是就可以称作为“原始设备映射磁盘”呢?这也是一种可以热...

大别阿郎
今天
9
0
【AngularJS学习笔记】02 小杂烩及学习总结

本文转载于:专业的前端网站☞【AngularJS学习笔记】02 小杂烩及学习总结 表格示例 <div ng-app="myApp" ng-controller="customersCtrl"> <table> <tr ng-repeat="x in names | orderBy ......

前端老手
昨天
13
0
Linux 内核的五大创新

在科技行业,创新这个词几乎和革命一样到处泛滥,所以很难将那些夸张的东西与真正令人振奋的东西区分开来。Linux内核被称为创新,但它又被称为现代计算中最大的奇迹,一个微观世界中的庞然大...

阮鹏
昨天
16
0
【Medium 万赞好文】ViewModel 和 LIveData:模式 + 反模式

原文作者: Jose Alcérreca 原文地址: ViewModels and LiveData: Patterns + AntiPatterns 译者:秉心说 View 和 ViewModel 分配责任 理想情况下,ViewModel 应该对 Android 世界一无所知。...

秉心说
昨天
17
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部