Spark-RDD操作Hbase

我不是女神ヾ 2023-02-18 14:30 71阅读 0赞

由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat 类的实现，Spark 可以通过Hadoop输入格式访问HBase。这个输入格式会返回键值对数据，其中键的类型为org. apache.hadoop.hbase.io.ImmutableBytesWritable，而值的类型为org.apache.hadoop.hbase.client.  
Result。  
（1）添加依赖

<dependency>
    	<groupId>org.apache.hbase</groupId>
    	<artifactId>hbase-server</artifactId>
    	<version>1.3.1</version>
    </dependency>
    
    <dependency>
    	<groupId>org.apache.hbase</groupId>
    	<artifactId>hbase-client</artifactId>
    	<version>1.3.1</version>
    </dependency>

把hbase中conf下的hbase-site.xml复制到idea中的resources  
**`名字必须要一样`**  
![在这里插入图片描述][20200618153927555.png]  
首先在hbase创建一张表

create 'student' ,'info'
    put 'student','1001','info:name','zhangsan'

代码如下

import org.apache.hadoop.hbase.{ CellUtil, HBaseConfiguration}
    import org.apache.hadoop.hbase.client.{ Put, Result}
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapred.TableOutputFormat
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.hadoop.mapred.JobConf
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Spark_Hbase { 
      def main(args: Array[String]): Unit = { 
    
        val config = new SparkConf().setMaster("local[*]").setAppName("practice")
        val sc = new SparkContext(config)
    
        val conf = HBaseConfiguration.create()
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104")
        conf.set(TableInputFormat.INPUT_TABLE, "student")
    
        //查询表
      val hbaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
          classOf[ImmutableBytesWritable],
          classOf[Result]
        )
        hbaseRDD.foreach { 
          case (rowkey, result) => { 
            val cells = result.rawCells()
            for (cell <- cells) { 
              println(Bytes.toString(CellUtil.cloneValue(cell)))
            }
          }
        }
        //插入表
      val dataRDD = sc.makeRDD(List(("1002", "zhangsan"), ("1003", "lisi"),
        ("1004", "wangwu")))
    
        val putRDD = dataRDD.map { 
          case (rowkey, name) => { 
            val put = new Put(Bytes.toBytes(rowkey))
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"),
              Bytes.toBytes(name))
            (new ImmutableBytesWritable(Bytes.toBytes(rowkey)), put)
          }
        }
        val jobConf = new JobConf(conf)
        jobConf.setOutputFormat(classOf[TableOutputFormat])
        jobConf.set(TableOutputFormat.OUTPUT_TABLE, "student")
    
        putRDD.saveAsHadoopDataset(jobConf)
    
        sc.stop()
      }
    
    
    }

[20200618153927555.png]: https://img-blog.csdnimg.cn/20200618153927555.png