(转)Scala操作Hbase

import java.io.IOException
import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HConstants, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.mutable.ListBuffer
/**
  * 从hbase中增删改查数据
  *
  */
object HbaseUtils {
  var zookeeperQuorum = "根据自己zookeeper情况"
  val configuration = HBaseConfiguration.create()
  configuration.set(HConstants.ZOOKEEPER_QUORUM, zookeeperQuorum)
  configuration.set ("zookeeper.znode.parent", "/hbase-unsecure") //看情况有时候要加有时候不加
  val connection = ConnectionFactory.createConnection(configuration)
  val admin = connection.getAdmin

  def isExists(tableName: String):Boolean ={
    var result=false
    val tName = TableName.valueOf(tableName)
    if(admin.tableExists(tName)) {
        result=true
      }
    result
  }
  //创建一个hbase表
  def createTable(tableName: String, columnFamilys: Array[String]) = {
      //操作的表名
      val tName = TableName.valueOf(tableName)
      //当表不存在的时候创建Hbase表
      if (!admin.tableExists(tName)) {
        //创建Hbase表模式
        val descriptor = new HTableDescriptor(tName)
        //创建列簇i
        for (columnFamily <- columnFamilys) {
          descriptor.addFamily(new HColumnDescriptor(columnFamily))
        }
        //创建表
        admin.createTable(descriptor)
        println("create successful!!")
      }
  }

  def dropTable(tableName: String):Unit={
      admin.disableTable(TableName.valueOf(tableName))
      admin.deleteTable(TableName.valueOf(tableName))
      println("drop successful!!")
  }
  //向hbase表中插入数据
  //put 'sk:test1','1','i:name','Luck2'
  def insertTable(tableName:String,rowkey: String, columnFamily: String, column: String, value: String) = {
      val table = connection.getTable(TableName.valueOf(tableName))
      //准备key 的数据
      val puts = new Put(rowkey.getBytes())
      //添加列簇名,字段名,字段值value
      puts.addColumn(columnFamily.getBytes(), column.getBytes(), value.getBytes())
      //把数据插入到tbale中
      table.put(puts)
      println("insert successful!!")
  }

  //获取hbase表中的数据
  //scan 'sk:test1'
  def scanDataFromHTable(tableName:String,columnFamily: String, column: String) = {
      val table = connection.getTable(TableName.valueOf(tableName))
      //定义scan对象
      val scan = new Scan()
      //添加列簇名称
      scan.addFamily(columnFamily.getBytes())
      //从table中抓取数据来scan
      val scanner = table.getScanner(scan)
      var result = scanner.next()
      //数据不为空时输出数据
      while (result != null) {
        println(s"rowkey:${Bytes.toString(result.getRow)},列簇:${columnFamily}:${column},value:${Bytes.toString(result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(column)))}")
        result = scanner.next()
      }
      //通过scan取完数据后,记得要关闭ResultScanner,否则RegionServer可能会出现问题(对应的Server资源无法释放)
      scanner.close()
  }
//只传表名得到全表数据
def getAllData(tableName: String): ListBuffer[String] = {
    var table: Table = null
    val list=new ListBuffer[String]
    try {
      table = connection.getTable(TableName.valueOf(tableName))
      val results: ResultScanner = table.getScanner(new Scan)
      import scala.collection.JavaConversions._
      for (result <- results) {
        for (cell <- result.rawCells) {
          val row: String = Bytes.toString(cell.getRowArray, cell.getRowOffset, cell.getRowLength)
          val family: String = Bytes.toString(cell.getFamilyArray, cell.getFamilyOffset, cell.getFamilyLength)
          val colName: String = Bytes.toString(cell.getQualifierArray, cell.getQualifierOffset, cell.getQualifierLength)
          val value: String = Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)
          val context: String = "rowkey:" + row +","+ "列族:" + family +","+ "列:" + colName +","+ "值:" + value
          list+=context
        }
      }
      results.close()
    } catch {
      case e: IOException =>
        e.printStackTrace()
    }
    list
  }

  //删除某条记录
  //delete 'sk:test1','1','i:name'
  def deleteRecord(tableName:String,rowkey: String, columnFamily: String, column: String) = {
      val table = connection.getTable(TableName.valueOf(tableName))

      val info = new Delete(Bytes.toBytes(rowkey))
      info.addColumn(columnFamily.getBytes(), column.getBytes())
      table.delete(info)
      println("delete successful!!")
  }
  // 关闭 connection 连接
  def close()={
    if (connection!=null){
      try{
        connection.close()
        println("关闭成功!")
      }catch{
        case e:IOException => println("关闭失败!")
      }
    }
  }

  def main(args: Array[String]): Unit = {
    /*var arr=new Array[String](1)
    arr(0)="info1"
    createTable("user2",arr)*/
    //insertTable("user2","1","info1","name","lyh")
    //scanDataFromHTable("user2","info1","name")
    //deleteRecord("user2","1","info1","name")
    dropTable("user2")
    println(isExists("user2"))
    close()
  }
}

转自:https://www.jianshu.com/p/0e0d6c7d8a42

Scala 中使用 HBase 和 Kerberos 进行身份验证需要进行以下步骤: 1. 配置 Kerberos:首先,确保 HBase 集群已经启用了 Kerberos 身份验证。您需要获取正确的 Kerberos Principal(主体)和 Keytab(密钥表)文件。 2. 添加 HBase 和 Kerberos依赖:在 Scala项目的构建文件(例如 build.sbt)中添加必要的依赖项。例如,您可以添加以下依赖项: ```scalalibraryDependencies += "org.apache.hbase" % "hbase-client" % "版本号" libraryDependencies += "org.apache.hadoop" % "hadoop-auth" % "版本号" ``` 请确保将版本号替换为您正在使用的 HBase 和 Hadoop 版本。 3. 创建 HBase 配置:在 Scala代码中,您需要创建一个 HBase 配置对象,并设置相应的属性,以使用 Kerberos 进行身份验证。例如: ```scalaimport org.apache.hadoop.conf.Configurationimport org.apache.hadoop.security.UserGroupInformationval conf = HBaseConfiguration.create() conf.set("hbase.zookeeper.quorum", "zkQuorum") conf.set("hbase.zookeeper.property.clientPort", "zkPort") conf.set("zookeeper.znode.parent", "/hbase") conf.set("hadoop.security.authentication", "kerberos") // 设置 Kerberos Principal 和 Keytab 文件路径val krbPrincipal = "your_kerberos_principal" val krbKeytab = "path_to_your_keytab_file" System.setProperty("java.security.krb5.conf", "/etc/krb5.conf") UserGroupInformation.setConfiguration(conf) UserGroupInformation.loginUserFromKeytab(krbPrincipal, krbKeytab) ``` 请确保将 "zkQuorum" 和 "zkPort" 替换为您的 ZooKeeper 地址和端口。 4. 使用 HBase 进行操作:现在,您可以使用 ScalaHBase 客户端库进行各种操作,例如读取和写入数据。根据您的需求,使用适当的 HBase API 进行操作。 请注意,这只是一个简单的示例,您可能需要针对您的特定环境和需求进行更多的配置和调整。 希望这可以帮助您在 Scala 中使用 HBase 和 Kerberos 进行身份验证!如果您有任何进一步的问题,请随时提问。
评论 1
添加红包

请填写红包祝福语或标题

红包个数最小为10个

红包金额最低5元

当前余额3.43前往充值 >
需支付:10.00
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包
实付
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明:

1.余额是钱包充值的虚拟货币,按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载,可以购买VIP、付费专栏及课程。

余额充值