我上一篇寫了關於如何在spark中直接訪問有kerberos的hbase，現在我們需要對hbase進行全表的分散式掃描，在無kerberos的情況下通過sparkcontext的newApiHadoopRDD就可以達到目的，但有了kerberos的限制，這個方法就不行了，也許有人會想到通過我之前提到的ugi doAs方法解決，但是分散式掃描的情況下如果將newApiHadoopRDD放到doAs中，只能對當前節點起到作用，無法為其他節點賦予許可權，使用的時候仍然會出現證書無效的問題。

那麼，如何解決此問題呢 ？首先，儘量保證spark執行在yarn環境中，不要用stdandalone方式，接著就是如何許可權問題了，網上給出的比較多的方式是通過token，不過我沒有嘗試成功，cloudera提供了一種解決此問題的方法，就是過載RDD，自己實現一個newApiHadoopRDD。這樣只要將其中訪問hbase的部分放在ugi的doAs中就可以了。

該如何過載呢？RDD的過載並不困難，首先繼承RDD類，接著要實現getPartitions和compute兩個介面。

getPartitions是spark獲取分片資訊的方法。而compute則是用來獲取具體的資料。

下面來看程式碼：

import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.{ SparkContext, TaskContext }
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.SerializableWritable
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.security.Credentials
import org.apache.spark.rdd.RDD
import org.apache.spark.Partition
import org.apache.spark.InterruptibleIterator
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.Logging
import org.apache.spark.SparkHadoopMapReduceUtilExtended
import org.apache.hadoop.mapreduce.JobID
import org.apache.hadoop.io.Writable
import org.apache.hadoop.mapreduce.InputSplit
import java.text.SimpleDateFormat
import java.util.Date
import java.util.ArrayList
import org.apache.hadoop.security.UserGroupInformation
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod
import org.apache.hadoop.hbase.mapreduce.IdentityTableMapper
import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.security.UserGroupInformation
import java.security.PrivilegedExceptionAction
import java.security.PrivilegedAction
import org.apache.hadoop.mapreduce.RecordReader
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable


class HBaseScanRDD(sc: SparkContext,
                   usf: UserSecurityFunction,
                   @transient tableName: String,
                   @transient scan: Scan,
                   configBroadcast: Broadcast[SerializableWritable[Configuration]])
  extends RDD[(Array[Byte], java.util.List[(Array[Byte], Array[Byte], Array[Byte])])](sc, Nil)
  with SparkHadoopMapReduceUtilExtended
  with Logging {

  ///
  @transient val jobTransient = new Job(configBroadcast.value.value, "ExampleRead");
  if(usf.isSecurityEnable())
  {
    usf.login().doAs(new PrivilegedExceptionAction[Unit] {
      def run: Unit = 
          TableMapReduceUtil.initTableMapperJob(
    tableName, // input HBase table name
    scan, // Scan instance to control CF and attribute selection
    classOf[IdentityTableMapper], // mapper
    null, // mapper output key
    null, // mapper output value
    jobTransient)
    })
  }else{
    TableMapReduceUtil.initTableMapperJob(
    tableName, // input HBase table name
    scan, // Scan instance to control CF and attribute selection
    classOf[IdentityTableMapper], // mapper
    null, // mapper output key
    null, // mapper output value
    jobTransient)
  }
  
  

  @transient val jobConfigurationTrans = jobTransient.getConfiguration()
  jobConfigurationTrans.set(TableInputFormat.INPUT_TABLE, tableName)
  val jobConfigBroadcast = sc.broadcast(new SerializableWritable(jobConfigurationTrans))
  ////

  private val jobTrackerId: String = {
    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
    formatter.format(new Date())
  }

  @transient protected val jobId = new JobID(jobTrackerId, id)

  override def getPartitions: Array[Partition] = {

    //addCreds

    val tableInputFormat = new TableInputFormat
    tableInputFormat.setConf(jobConfigBroadcast.value.value)

    val jobContext = newJobContext(jobConfigBroadcast.value.value, jobId)
    var rawSplits : Array[Object] = null
    if(usf.isSecurityEnable())
    {
      rawSplits = usf.login().doAs(new PrivilegedAction[Array[Object]]{
      def run: Array[Object] = tableInputFormat.getSplits(jobContext).toArray
    })
    }else{
      rawSplits = tableInputFormat.getSplits(jobContext).toArray
    }
      
    
    val result = new Array[Partition](rawSplits.size)
    for (i <- 0 until rawSplits.size) {
      result(i) = new NewHadoopPartition(id, i, rawSplits(i).asInstanceOf[InputSplit with Writable])
    }

    result
  }

  override def compute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(Array[Byte], java.util.List[(Array[Byte], Array[Byte], Array[Byte])])] = {

    //addCreds

    val iter = new Iterator[(Array[Byte], java.util.List[(Array[Byte], Array[Byte], Array[Byte])])] {

      //addCreds

      val split = theSplit.asInstanceOf[NewHadoopPartition]
      logInfo("Input split: " + split.serializableHadoopSplit)
      val conf = jobConfigBroadcast.value.value

      val attemptId = newTaskAttemptID(jobTrackerId, id, isMap = true, split.index, 0)
      val hadoopAttemptContext = newTaskAttemptContext(conf, attemptId)
      val format = new TableInputFormat
      format.setConf(conf)

      var reader : RecordReader[ImmutableBytesWritable, Result] = null
      if(usf.isSecurityEnable())
      {
      reader = usf.login().doAs(new PrivilegedAction[RecordReader[ImmutableBytesWritable, Result]]{
        def run: RecordReader[ImmutableBytesWritable, Result] = {
         val _reader = format.createRecordReader(
        split.serializableHadoopSplit.value, hadoopAttemptContext)
      _reader.initialize(split.serializableHadoopSplit.value, hadoopAttemptContext)
      _reader
        }
      })}else{
        reader = format.createRecordReader(
        split.serializableHadoopSplit.value, hadoopAttemptContext)
        reader.initialize(split.serializableHadoopSplit.value, hadoopAttemptContext)
      }
       

      // Register an on-task-completion callback to close the input stream.
      context.addOnCompleteCallback(() => close())
      var havePair = false
      var finished = false

      override def hasNext: Boolean = {
        if (!finished && !havePair) {
          finished = !reader.nextKeyValue
          havePair = !finished
        }
        !finished
      }

      override def next(): (Array[Byte], java.util.List[(Array[Byte], Array[Byte], Array[Byte])]) = {
        if (!hasNext) {
          throw new java.util.NoSuchElementException("End of stream")
        }
        havePair = false

        val it = reader.getCurrentValue.listCells().iterator()

        val list = new ArrayList[(Array[Byte], Array[Byte], Array[Byte])]()

        while (it.hasNext()) {
          val kv = it.next()
          list.add((CellUtil.cloneFamily(kv), CellUtil.cloneQualifier(kv), CellUtil.cloneValue(kv)))
        }
        (reader.getCurrentKey.copyBytes(), list)
      }

      private def close() {
        try {
          reader.close()
        } catch {
          case e: Exception => logWarning("Exception in RecordReader.close()", e)
        }
      }
    }
    new InterruptibleIterator(context, iter)
  }

  def addCreds {
    val creds = SparkHadoopUtil.get.getCurrentUserCredentials()

    val ugi = UserGroupInformation.getCurrentUser();
    ugi.addCredentials(creds)
    // specify that this is a proxy user 
    ugi.setAuthenticationMethod(AuthenticationMethod.PROXY)
  }

  private class NewHadoopPartition(
                                           rddId: Int,
                                           val index: Int,
                                           @transient rawSplit: InputSplit with Writable)
    extends Partition {

    val serializableHadoopSplit = new SerializableWritable(rawSplit)

    override def hashCode(): Int = 41 * (41 + rddId) + index
  }
}
 

package org.apache.spark

import org.apache.spark.mapreduce.SparkHadoopMapReduceUtil

trait SparkHadoopMapReduceUtilExtended extends SparkHadoopMapReduceUtil{
  
}

這樣就可以直接使用，new HBaseScanRDD()，傳入SparkContext。

如果想用java訪問，還需要在外面封裝一層。

import org.apache.spark.api.java.JavaSparkContext
import org.apache.hadoop.conf.Configuration
import org.apache.spark.broadcast.Broadcast
import org.apache.hadoop.hbase.client.Scan
import org.apache.spark.SerializableWritable
import org.apache.spark.api.java.JavaPairRDD

class JavaHBaseContext(@transient jsc: JavaSparkContext,usf: UserSecurityFunction) {
  
  def hbaseApiRDD(tableName: String,
      scan: Scan,
      configBroadcast: Configuration) = {
      JavaPairRDD.fromRDD(new HBaseScanRDD(JavaSparkContext.toSparkContext(jsc),usf,tableName,
        scan,
        jsc.broadcast(new SerializableWritable(configBroadcast))))
  }
  
}