ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-6918] [YARN] Secure HBase support.

Browse source 
Obtain HBase security token with Kerberos credentials locally to be sent to executors. Tested on eBay's secure HBase cluster.

Similar to obtainTokenForNamenodes and fails gracefully if HBase classes are not included in path.

Requires hbase-site.xml to be in the classpath(typically via conf dir) for the zookeeper configuration. Should that go in the docs somewhere? Did not see an HBase section.

Author: Dean Chen <deanchen5@gmail.com>

Closes #5586 from deanchen/master and squashes the following commits:

0c190ef [Dean Chen] [SPARK-6918][YARN] Secure HBase support.
This commit is contained in:
Dean Chen 2015-04-29 08:58:33 -05:00 committed by Thomas Graves
parent f49284b5bf
commit baed3f2c73
1 changed files with 37 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
View file

@ -39,7 +39,7 @@ import org.apache.hadoop.io.Text
import org.apache.hadoop.mapred.Master
import org.apache.hadoop.mapreduce.MRJobConfig
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.hadoop.security.token.Token
import org.apache.hadoop.security.token.{TokenIdentifier, Token}
import org.apache.hadoop.util.StringUtils
import org.apache.hadoop.yarn.api._
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment
@ -226,6 +226,7 @@ private[spark] class Client(
val distributedUris = new HashSet[String]
obtainTokensForNamenodes(nns, hadoopConf, credentials)
obtainTokenForHiveMetastore(hadoopConf, credentials)
obtainTokenForHBase(hadoopConf, credentials)
val replication = sparkConf.getInt("spark.yarn.submit.file.replication",
fs.getDefaultReplication(dst)).toShort
@ -1084,6 +1085,41 @@ object Client extends Logging {
}
}
/**
* Obtain security token for HBase.
*/
def obtainTokenForHBase(conf: Configuration, credentials: Credentials): Unit = {
if (UserGroupInformation.isSecurityEnabled) {
val mirror = universe.runtimeMirror(getClass.getClassLoader)
try {
val confCreate = mirror.classLoader.
loadClass("org.apache.hadoop.hbase.HBaseConfiguration").
getMethod("create", classOf[Configuration])
val obtainToken = mirror.classLoader.
loadClass("org.apache.hadoop.hbase.security.token.TokenUtil").
getMethod("obtainToken", classOf[Configuration])
logDebug("Attempting to fetch HBase security token.")
val hbaseConf = confCreate.invoke(null, conf)
val token = obtainToken.invoke(null, hbaseConf).asInstanceOf[Token[TokenIdentifier]]
credentials.addToken(token.getService, token)
logInfo("Added HBase security token to credentials.")
} catch {
case e:java.lang.NoSuchMethodException =>
logInfo("HBase Method not found: " + e)
case e:java.lang.ClassNotFoundException =>
logDebug("HBase Class not found: " + e)
case e:java.lang.NoClassDefFoundError =>
logDebug("HBase Class not found: " + e)
case e:Exception =>
logError("Exception when obtaining HBase security token: " + e)
}
}
}
/**
* Return whether the two file systems are the same.
*/