Hadoop集群交互用户的永久Kerberos票证

--principal --keytab

https://spark.apache.org/docs/latest/security.html#yarn-mode

分别通过--principal和--keytab参数。钥匙扣 通过Hadoop分布式缓存进行主控(如果纱线

您可以在Spark Driver日志中看到,当yarn更新kerberos票据时。

val configuration = new Configuration
configuration.addResource("/etc/hadoop/conf/hdfs-site.xml")
UserGroupInformation.setConfiguration(configuration)

UserGroupInformation.getCurrentUser.setAuthenticationMethod(AuthenticationMethod.KERBEROS)
UserGroupInformation.loginUserFromKeytabAndReturnUGI(
  "hadoop.kerberos.principal", " path of hadoop.kerberos.keytab file")
  .doAs(new PrivilegedExceptionAction[Unit]() {
    @Override
    def run(): Unit = {
       //hbase/hive connection
      // logic

    }
  })

如果除了spark没有其他组件访问权限,则不需要编写上述代码。只需在spark submit命令中提供keytab和principal。

spark-submit --master yarn-cluster --keytab "xxxxxx.keytab" --principal "svc-xxxx@xxxx.COM"  xxxx.jar

--键选项卡

#!/usr/bin/python2.7

from __future__ import print_function

import os
import sys
import stat
import getpass
import subprocess
import collections
import socket
import tempfile

def runSudo(cmd, pw):
    try:
        subprocess.check_call("echo '{}' | sudo -S -p '' {}".format(pw, cmd), shell = True)
        return True
    except subprocess.CalledProcessError:
        return False

def testPassword(pw):
    subprocess.check_call("sudo -k", shell = True)
    if not runSudo("true", pw):
        print("Incorrect password for user {}".format(getpass.getuser()), file = sys.stderr)
        sys.exit(os.EX_NOINPUT)    

class KeytabFile(object):
    def __init__(self, pw):
        self.userName = getpass.getuser()
        self.pw = pw
        self.targetPath = "/etc/security/keytabs/{}.headless.keytab".format(self.userName)
        self.tempFile = None

    KeytabEntry = collections.namedtuple("KeytabEntry", ("kvno", "principal", "encryption"))

    def LoadExistingKeytab(self):
        if not os.access(self.targetPath, os.R_OK):

            # Note: the assumption made here, that the Kerberos realm is same as the DNS domain,
            # may not hold in other setups
            domainName = ".".join(socket.getfqdn().split(".")[1:])

            encryptions = ("aes128-cts-hmac-sha1-96", "arcfour-hmac", "aes256-cts-hmac-sha1-96")
            return [
                self.KeytabEntry(0, "@".join( (self.userName, domainName)), encryption)
                    for encryption in encryptions ]

        def parseLine(keytabLine):
            tokens = keytabLine.strip().split(" ")
            return self.KeytabEntry(int(tokens[0]), tokens[1], tokens[2].strip("()"))

        cmd ="klist -ek {} | tail -n+4".format(self.targetPath)
        entryLines = subprocess.check_output(cmd, shell = True).splitlines()
        return map(parseLine, entryLines)

    class KtUtil(subprocess.Popen):
        def __init__(self):
            subprocess.Popen.__init__(self, "ktutil",
                stdin = subprocess.PIPE, stdout = subprocess.PIPE, stderr=subprocess.PIPE, shell = True)

        def SendLine(self, line, expectPrompt = True):
            self.stdin.write(bytes(line + "\n"))
            self.stdin.flush()
            if expectPrompt:
                self.stdout.readline()

        def Quit(self):
            self.SendLine("quit", False)
            rc = self.wait()
            if rc != 0:
                raise subprocess.CalledProcessError(rc, "ktutil")


    def InstallUpdatedKeytab(self):
        fd, tempKt = tempfile.mkstemp(suffix = ".keytab")
        os.close(fd)
        entries = self.LoadExistingKeytab()
        ktutil = self.KtUtil()
        for entry in entries:
            cmd = "add_entry -password -p {} -k {} -e {}".format(
                entry.principal, entry.kvno + 1, entry.encryption)

            ktutil.SendLine(cmd)
            ktutil.SendLine(self.pw)

        os.unlink(tempKt)
        ktutil.SendLine("write_kt {}".format(tempKt))
        ktutil.Quit()

        if not runSudo("mv {} {}".format(tempKt, self.targetPath), self.pw):
            os.unlink(tempKt)
            print("Failed to install the keytab to {}.".format(self.targetPath), file = sys.stderr)
            sys.exit(os.EX_CANTCREAT)

        os.chmod(self.targetPath, stat.S_IRUSR)
        # TODO: Also change group to 'hadoop'

if __name__ == '__main__':

    def main():
        userPass = getpass.getpass("Please enter your password: ")
        testPassword(userPass)
        kt = KeytabFile(userPass)
        kt.InstallUpdatedKeytab()

    main()