前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Spark读写Hbase中的数据

Spark读写Hbase中的数据

作者头像
用户3003813
发布2018-09-06 13:57:58
1.6K0
发布2018-09-06 13:57:58
举报
文章被收录于专栏:个人分享个人分享
代码语言:javascript
复制
def main(args: Array[String])  {
    val sparkConf = new SparkConf().setMaster("local").setAppName("cocapp").set("spark.kryo.registrator", classOf[HBaseConfiguration].getName)
      .set("spark.executor.memory", "4g")
    val sc: SparkContext = new SparkContext(sparkConf)
    val sqlContext = new HiveContext(sc)
    val mySQLUrl = "jdbc:mysql://localhost:3306/yangsy?user=root&password=yangsiyi"
    val rows = sqlContext.jdbc(mySQLUrl, "person")
    val tableName = "spark"
    val columnFamily = "cf" //rows.first().getString(1)
    val configuration = HBaseConfiguration.create();
    configuration.set(TableInputFormat.INPUT_TABLE, "test");
    val admin = new HBaseAdmin(configuration)
    val hBaseRDD = sc.newAPIHadoopRDD(configuration, classOf[TableInputFormat],
        classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
    classOf[org.apache.hadoop.hbase.client.Result])
      hBaseRDD.count()
代码语言:javascript
复制
def toHbase(rows: DataFrame,tableName : String,columnFamily: String)   {
    val configuration = HBaseConfiguration.create();
    val admin = new HBaseAdmin(configuration)
    if (admin.tableExists(tableName)) {
      print("table Exists")
      admin.disableTable(tableName);
      admin.deleteTable(tableName);
    }
    configuration.addResource("hbase-site.xml")
    val tableDesc = new HTableDescriptor(tableName)
    tableDesc.addFamily(new HColumnDescriptor(columnFamily))
    admin.createTable(tableDesc)
    rows.foreachPartition { row =>
      val table = new HTable(configuration, tableName)

      row.foreach { a =>
        val put = new Put(Bytes.toBytes("row1"))
        put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("coulumn1"), Bytes.toBytes(a.getString(0)))
        table.put(put)
        println("insert into success")
      }
    }

然而并没有什么乱用,发现一个问题,就是说,在RDD取值与写入HBASE的时候,引入外部变量无法序列化。。。。。。网上很多说法是说extends Serializable ,可是尝试无效。Count()是可以获取到,但是如果我要在configuration中set列,然后进行查询就会报错了。暂时各种办法尝试无果,还在想办法,也不明原因。

代码语言:javascript
复制
本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2015-12-02 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档