spark-submit cluster模式业务开发 --解决配置文件读取不到、安全认证不 ...

篮之新喜 · 9 小时前

个人主页：500佰
个人专栏：大数据组件性能调优-登上高阶
#spark #spark-submit #spark-sql #spark业务开发 #大数据 #生产环境真实案例 #整理 #履历总结
阐明：本专题：spark-submit cluster模式业务开发 含完备代码
业务场景：spark-sql类型任务

本案例：可以解决配置文件读取不到(如keytab、krb5.conf、spark配置文件)、安全认证不过等题目
本案例目标：使用spark-submit cluster模式提交spark-sql类型任务

cluster 模式优势：
相对client模式，driver端在集群内部，资源更好管理，减少垮节点IO
cluster模式常见题目：
相对client模式，存在读取不到配置文件及安全认证不过 √
本案例：可以解决配置文件读取不到、安全认证不过等题目
调用slala-sdk版本要求：2.12.14

调用示例:

userSparkSQL.sh 示例如下：

#!/bin/bash
function envInit(){
spark_submit_path=$(dirname "${BASH_SOURCE-$0}")
spark_submit_path=$(cd "$spark_submit_path"; pwd)
client=$(cd "$spark_submit_path"/../../../;pwd)
source "$client"/bigdata_env
}
envInit
${spark_submit_path}/spark-submit \
--master yarn \
--deploy-mode cluster \
--conf spark.sql.autoBroadcastJoinThreshold=-1 \
--conf spark.driver.memory=50g \
--conf spark.logOptimization.enable=false \
--conf spark.sql.shuffle.partitions=400 \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf spark.network.timeout=300 \
--num-executors 12 \
--executor-memory 6g \
--executor-cores 10 \
--name ExecSqlc\
--files /opt/client2/Spark2x/spark/bin/conf/user.keytab,/opt/client2/Spark2x/spark/bin/conf/krb5.conf,/opt/client2/Spark2x/spark/bin/conf/2.hql \
--class com.xx.spark.security.bigdata.ExecSqlc /opt/client2/Spark2x/spark/bin/userSparkSQL.jar \
--userPrincipal=hcqs \
--sparkSqlPath=/opt/client2/Spark2x/spark/bin/conf/2.hql \
--params="cloumn1=0201,cloumn2=2059881123"

复制代码

Hive SQL文件(示例的x.hql文件）内里的参数示例：

select * from tb where cloumn1=$cloumn1 and cloumn2=$cloumn2

复制代码

业务代码

主类 object ExecSqlc:

package com.mrs.spark.security.bigdata
import com.mrs.spark.security.hadoop.LoginUtil
import org.apache.log4j.Logger
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.storage.StorageLevel
import java.io.File
import scala.io.Source
/**
* 执行spark-sql类
* spark-submit driver cluster
*
* @param userPrincipal ,userKeytabPath,krb5ConfPath
* @return
* @author 500佰
*/
object ExecSqlc {
val logger = LoginUtil.LOG
// Table structure
def readSparkSQLFile(sparkSqlPath: String, variables: Map[String, String], sparkSession: SparkSession): String = {
val source = Source.fromFile(sparkSqlPath) // driver client model
try {
val sqlString = source.getLines().map { line =>
variables.foldLeft(line) { (acc, entry) =>
acc.replaceAllLiterally(s"$$${entry._1}", entry._2)
}
}.mkString("\n")
sqlString
} finally {
source.close()
}
}
def main(args: Array[String]) {
val logger = Logger.getLogger(getClass.getName)
val paramMap = args.map(_.split("=", 2)).collect {
case Array(key, value) if key.startsWith("--") => key.stripPrefix("--") -> value.trim
}.toMap
val userPrincipal = paramMap.getOrElse("userPrincipal", "")
// val userKeytabPath = paramMap.getOrElse("userKeytabPath", "")
// val krb5ConfPath = paramMap.getOrElse("krb5ConfPath", "")
val sparkSqlPath = paramMap.getOrElse("sparkSqlPath", "")
val filePath = System.getProperty("user.dir") + File.separator + sparkSqlPath.split("/").last
val params = paramMap.getOrElse("params", "").split(",").map(_.split("=")).map(arr => arr(0) -> arr(1)).toMap
// try {
LoginUtil.loginWithUserKeytab(userPrincipal)
logger.info("认证成功！")
// Configure SparkSession & Spark application name
val appName = sparkSqlPath.split("/").last.split("\\.")(0)
val spark = SparkSession
.builder()
.appName(s"ExecUserSqlc-${appName}")
.getOrCreate()
//sparkSql Content
val sqlContent = readSparkSQLFile(filePath, params, spark)
val sqlc: String = sqlContent.stripMargin //format
logger.info(s"执行语句：${sqlc}")
logger.info(s"开始执行：${sparkSqlPath}")
val sqlStatements: Array[String] = sqlc.split(";")
//遍历顺序执行
for (statement <- sqlStatements) {
val Info: DataFrame = spark.sql(statement).distinct().persist(StorageLevel.MEMORY_AND_DISK)
logger.info(s"执行结果：${Info.show(2)}")
}
spark.stop()
// System.exit(0)
// } catch {
// case ex: Exception => logger.error("发生了异常：" + ex.getMessage)
// System.exit(1)
// }
}
}

复制代码

其他代码：

com.xx.spark.security.hadoop.KerberosUtil类：

package com.mrs.spark.security.hadoop;
import org.apache.log4j.Logger;
import java.lang.reflect.Method;
public class KerberosUtil {
private static Logger logger = Logger.getLogger(KerberosUtil.class);
public static final String JAVA_VENDER = "java.vendor";
public static final String IBM_FLAG = "IBM";
public static final String CONFIG_CLASS_FOR_IBM = "com.ibm.security.krb5.internal.Config";
public static final String CONFIG_CLASS_FOR_SUN = "sun.security.krb5.Config";
public static final String METHOD_GET_INSTANCE = "getInstance";
public static final String METHOD_GET_DEFAULT_REALM = "getDefaultRealm";
public static final String DEFAULT_REALM = "HADOOP.COM";
// UUID principal
// principal=spark2x/hadoop.868023f3-49f9-4be4-ae96-bc723d246362.com@868023F3-49F9-4BE4-AE96-BC723D246362.COM
// Default principal principal=spark2x/hadoop.hadoop.com@HADOOP.COM
public static String getKrb5DomainRealm() {
Class<?> krb5ConfClass;
String peerRealm;
try {
if (System.getProperty(JAVA_VENDER).contains(IBM_FLAG)) {
krb5ConfClass = Class.forName(CONFIG_CLASS_FOR_IBM);
} else {
krb5ConfClass = Class.forName(CONFIG_CLASS_FOR_SUN);
}
Method getInstanceMethod = krb5ConfClass.getMethod(METHOD_GET_INSTANCE);
Object kerbConf = getInstanceMethod.invoke(krb5ConfClass);
Method getDefaultRealmMethod = krb5ConfClass.getDeclaredMethod(METHOD_GET_DEFAULT_REALM);
peerRealm = (String) getDefaultRealmMethod.invoke(kerbConf);
logger.info("Get default realm successfully, the realm is : " + peerRealm);
} catch (Exception e) {
peerRealm = DEFAULT_REALM;
logger.warn("Get default realm failed, use default value : " + DEFAULT_REALM);
}
return peerRealm;
}
}

复制代码

com.xx.spark.security.hadoop.LoginUtil类：

package com.mrs.spark.security.hadoop;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authentication.util.KerberosUtil;
import org.apache.log4j.Logger;
import javax.security.auth.login.AppConfigurationEntry;
import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class LoginUtil {
public enum Module {
STORM("StormClient"),
KAFKA("KafkaClient"),
ZOOKEEPER("Client");
private String name;
private Module(String name) {
this.name = name;
}
public String getName() {
return name;
}
}
public static final Logger LOG = Logger.getLogger(LoginUtil.class);
/**
* line operator string
*/
private static final String LINE_SEPARATOR = System.getProperty("line.separator");
/**
* jaas file postfix
*/
private static final String JAAS_POSTFIX = ".jaas.conf";
/**
* IBM jdk login module
*/
private static final String IBM_LOGIN_MODULE = "com.ibm.security.auth.module.Krb5LoginModule required";
/**
* oracle jdk login module
*/
private static final String SUN_LOGIN_MODULE = "com.sun.security.auth.module.Krb5LoginModule required";
/**
* java security login file path
*/
public static final String JAVA_SECURITY_LOGIN_CONF_KEY = "java.security.auth.login.config";
private static final String JAVA_SECURITY_KRB5_CONF_KEY = "java.security.krb5.conf";
private static final String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
private static final String LOGIN_FAILED_CAUSE_PASSWORD_WRONG =
"(wrong password) keytab file and user not match, you can kinit -k -t keytab user in client server to"
+ " check";
private static final String LOGIN_FAILED_CAUSE_TIME_WRONG =
"(clock skew) time of local server and remote server not match, please check ntp to remote server";
private static final String LOGIN_FAILED_CAUSE_AES256_WRONG =
"(aes256 not support) aes256 not support by default jdk/jre, need copy local_policy.jar and"
+ " US_export_policy.jar from remote server in path /opt/huawei/Bigdata/jdk/jre/lib/security";
private static final String LOGIN_FAILED_CAUSE_PRINCIPAL_WRONG =
"(no rule) principal format not support by default, need add property hadoop.security.auth_to_local(in"
+ " core-site.xml) value RULE:[1:$1] RULE:[2:$1]";
private static final String LOGIN_FAILED_CAUSE_TIME_OUT =
"(time out) can not connect to kdc server or there is fire wall in the network";
private static final boolean IS_IBM_JDK = System.getProperty("java.vendor").contains("IBM");
public static synchronized void login(
String userPrincipal, String userKeytabPath, String krb5ConfPath, Configuration conf) throws IOException {
// 1.check input parameters
if ((userPrincipal == null) || (userPrincipal.length() <= 0)) {
LOG.error("input userPrincipal is invalid.");
throw new IOException("input userPrincipal is invalid.");
}
if ((userKeytabPath == null) || (userKeytabPath.length() <= 0)) {
LOG.error("input userKeytabPath is invalid.");
throw new IOException("input userKeytabPath is invalid.");
}
if ((krb5ConfPath == null) || (krb5ConfPath.length() <= 0)) {
LOG.error("input krb5ConfPath is invalid.");
throw new IOException("input krb5ConfPath is invalid.");
}
if ((conf == null)) {
LOG.error("input conf is invalid.");
throw new IOException("input conf is invalid.");
}
// 2.check file exsits
File userKeytabFile = new File(userKeytabPath);
if (!userKeytabFile.exists()) {
LOG.error("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
throw new IOException("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
}
if (!userKeytabFile.isFile()) {
LOG.error("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") is not a file.");
throw new IOException("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") is not a file.");
}
File krb5ConfFile = new File(krb5ConfPath);
if (!krb5ConfFile.exists()) {
LOG.error("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") does not exsit.");
throw new IOException("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") does not exsit.");
}
if (!krb5ConfFile.isFile()) {
LOG.error("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") is not a file.");
throw new IOException("krb5ConfFile(" + krb5ConfFile.getAbsolutePath() + ") is not a file.");
}
// 3.set and check krb5config
setKrb5Config(krb5ConfFile.getAbsolutePath());
setConfiguration(conf);
// 4.login and check for hadoop
loginHadoop(userPrincipal, userKeytabFile.getAbsolutePath());
LOG.info("Login success!!!!!!!!!!!!!!");
}
private static void setConfiguration(Configuration conf) throws IOException {
UserGroupInformation.setConfiguration(conf);
}
private static boolean checkNeedLogin(String principal) throws IOException {
if (!UserGroupInformation.isSecurityEnabled()) {
LOG.error(
"UserGroupInformation is not SecurityEnabled, please check if core-site.xml exists in classpath.");
throw new IOException(
"UserGroupInformation is not SecurityEnabled, please check if core-site.xml exists in classpath.");
}
UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
if ((currentUser != null) && (currentUser.hasKerberosCredentials())) {
if (checkCurrentUserCorrect(principal)) {
LOG.info("current user is " + currentUser + "has logined.");
if (!currentUser.isFromKeytab()) {
LOG.error("current user is not from keytab.");
throw new IOException("current user is not from keytab.");
}
return false;
} else {
LOG.error(
"current user is "
+ currentUser
+ "has logined. please check your enviroment , especially when it used IBM JDK or"
+ " kerberos for OS count login!!");
throw new IOException(
"current user is " + currentUser + " has logined. And please check your enviroment!!");
}
}
return true;
}
public static void setKrb5Config(String krb5ConfFile) throws IOException {
System.setProperty(JAVA_SECURITY_KRB5_CONF_KEY, krb5ConfFile);
String ret = System.getProperty(JAVA_SECURITY_KRB5_CONF_KEY);
if (ret == null) {
LOG.error(JAVA_SECURITY_KRB5_CONF_KEY + " is null.");
throw new IOException(JAVA_SECURITY_KRB5_CONF_KEY + " is null.");
}
if (!ret.equals(krb5ConfFile)) {
LOG.error(JAVA_SECURITY_KRB5_CONF_KEY + " is " + ret + " is not " + krb5ConfFile + ".");
throw new IOException(JAVA_SECURITY_KRB5_CONF_KEY + " is " + ret + " is not " + krb5ConfFile + ".");
}
}
public static void setJaasFile(String principal, String keytabPath) throws IOException {
String jaasPath =
new File(System.getProperty("java.io.tmpdir"))
+ File.separator
+ System.getProperty("user.name")
+ JAAS_POSTFIX;
// windows路径下分隔符替换
jaasPath = jaasPath.replace("\", "\\\");
keytabPath = keytabPath.replace("\", "\\\");
// 删除jaas文件
deleteJaasFile(jaasPath);
writeJaasFile(jaasPath, principal, keytabPath);
System.setProperty(JAVA_SECURITY_LOGIN_CONF_KEY, jaasPath);
}
private static void writeJaasFile(String jaasPath, String principal, String keytabPath) throws IOException {
FileWriter writer = new FileWriter(new File(jaasPath));
try {
writer.write(getJaasConfContext(principal, keytabPath));
writer.flush();
} catch (IOException e) {
throw new IOException("Failed to create jaas.conf File");
} finally {
writer.close();
}
}
private static void deleteJaasFile(String jaasPath) throws IOException {
File jaasFile = new File(jaasPath);
if (jaasFile.exists()) {
if (!jaasFile.delete()) {
throw new IOException("Failed to delete exists jaas file.");
}
}
}
private static String getJaasConfContext(String principal, String keytabPath) {
Module[] allModule = Module.values();
StringBuilder builder = new StringBuilder();
for (Module modlue : allModule) {
builder.append(getModuleContext(principal, keytabPath, modlue));
}
return builder.toString();
}
private static String getModuleContext(String userPrincipal, String keyTabPath, Module module) {
StringBuilder builder = new StringBuilder();
if (IS_IBM_JDK) {
builder.append(module.getName()).append(" {").append(LINE_SEPARATOR);
builder.append(IBM_LOGIN_MODULE).append(LINE_SEPARATOR);
builder.append("credsType=both").append(LINE_SEPARATOR);
builder.append("principal="" + userPrincipal + """).append(LINE_SEPARATOR);
builder.append("useKeytab="" + keyTabPath + """).append(LINE_SEPARATOR);
builder.append("debug=true;").append(LINE_SEPARATOR);
builder.append("};").append(LINE_SEPARATOR);
} else {
builder.append(module.getName()).append(" {").append(LINE_SEPARATOR);
builder.append(SUN_LOGIN_MODULE).append(LINE_SEPARATOR);
builder.append("useKeyTab=true").append(LINE_SEPARATOR);
builder.append("keyTab="" + keyTabPath + """).append(LINE_SEPARATOR);
builder.append("principal="" + userPrincipal + """).append(LINE_SEPARATOR);
builder.append("useTicketCache=false").append(LINE_SEPARATOR);
builder.append("storeKey=true").append(LINE_SEPARATOR);
builder.append("debug=true;").append(LINE_SEPARATOR);
builder.append("};").append(LINE_SEPARATOR);
}
return builder.toString();
}
public static void setJaasConf(String loginContextName, String principal, String keytabFile) throws IOException {
if ((loginContextName == null) || (loginContextName.length() <= 0)) {
LOG.error("input loginContextName is invalid.");
throw new IOException("input loginContextName is invalid.");
}
if ((principal == null) || (principal.length() <= 0)) {
LOG.error("input principal is invalid.");
throw new IOException("input principal is invalid.");
}
if ((keytabFile == null) || (keytabFile.length() <= 0)) {
LOG.error("input keytabFile is invalid.");
throw new IOException("input keytabFile is invalid.");
}
File userKeytabFile = new File(keytabFile);
if (!userKeytabFile.exists()) {
LOG.error("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
throw new IOException("userKeytabFile(" + userKeytabFile.getAbsolutePath() + ") does not exsit.");
}
javax.security.auth.login.Configuration.setConfiguration(
new JaasConfiguration(loginContextName, principal, userKeytabFile.getAbsolutePath()));
javax.security.auth.login.Configuration conf = javax.security.auth.login.Configuration.getConfiguration();
if (!(conf instanceof JaasConfiguration)) {
LOG.error("javax.security.auth.login.Configuration is not JaasConfiguration.");
throw new IOException("javax.security.auth.login.Configuration is not JaasConfiguration.");
}
AppConfigurationEntry[] entrys = conf.getAppConfigurationEntry(loginContextName);
if (entrys == null) {
LOG.error(
"javax.security.auth.login.Configuration has no AppConfigurationEntry named "
+ loginContextName
+ ".");
throw new IOException(
"javax.security.auth.login.Configuration has no AppConfigurationEntry named "
+ loginContextName
+ ".");
}
boolean checkPrincipal = false;
boolean checkKeytab = false;
for (int i = 0; i < entrys.length; i++) {
if (entrys[i].getOptions().get("principal").equals(principal)) {
checkPrincipal = true;
}
if (IS_IBM_JDK) {
if (entrys[i].getOptions().get("useKeytab").equals(keytabFile)) {
checkKeytab = true;
}
} else {
if (entrys[i].getOptions().get("keyTab").equals(keytabFile)) {
checkKeytab = true;
}
}
}
if (!checkPrincipal) {
LOG.error(
"AppConfigurationEntry named "
+ loginContextName
+ " does not have principal value of "
+ principal
+ ".");
throw new IOException(
"AppConfigurationEntry named "
+ loginContextName
+ " does not have principal value of "
+ principal
+ ".");
}
if (!checkKeytab) {
LOG.error(
"AppConfigurationEntry named "
+ loginContextName
+ " does not have keyTab value of "
+ keytabFile
+ ".");
throw new IOException(
"AppConfigurationEntry named "
+ loginContextName
+ " does not have keyTab value of "
+ keytabFile
+ ".");
}
}
public static void setZookeeperServerPrincipal(String zkServerPrincipal) throws IOException {
System.setProperty(ZOOKEEPER_SERVER_PRINCIPAL_KEY, zkServerPrincipal);
String ret = System.getProperty(ZOOKEEPER_SERVER_PRINCIPAL_KEY);
if (ret == null) {
LOG.error(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is null.");
throw new IOException(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is null.");
}
if (!ret.equals(zkServerPrincipal)) {
LOG.error(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is " + ret + " is not " + zkServerPrincipal + ".");
throw new IOException(ZOOKEEPER_SERVER_PRINCIPAL_KEY + " is " + ret + " is not " + zkServerPrincipal + ".");
}
}
@Deprecated
public static void setZookeeperServerPrincipal(String zkServerPrincipalKey, String zkServerPrincipal)
throws IOException {
System.setProperty(zkServerPrincipalKey, zkServerPrincipal);
String ret = System.getProperty(zkServerPrincipalKey);
if (ret == null) {
LOG.error(zkServerPrincipalKey + " is null.");
throw new IOException(zkServerPrincipalKey + " is null.");
}
if (!ret.equals(zkServerPrincipal)) {
LOG.error(zkServerPrincipalKey + " is " + ret + " is not " + zkServerPrincipal + ".");
throw new IOException(zkServerPrincipalKey + " is " + ret + " is not " + zkServerPrincipal + ".");
}
}
private static void loginHadoop(String principal, String keytabFile) throws IOException {
try {
UserGroupInformation.loginUserFromKeytab(principal, keytabFile);
} catch (IOException e) {
LOG.error("login failed with " + principal + " and " + keytabFile + ".");
LOG.error("perhaps cause 1 is " + LOGIN_FAILED_CAUSE_PASSWORD_WRONG + ".");
LOG.error("perhaps cause 2 is " + LOGIN_FAILED_CAUSE_TIME_WRONG + ".");
LOG.error("perhaps cause 3 is " + LOGIN_FAILED_CAUSE_AES256_WRONG + ".");
LOG.error("perhaps cause 4 is " + LOGIN_FAILED_CAUSE_PRINCIPAL_WRONG + ".");
LOG.error("perhaps cause 5 is " + LOGIN_FAILED_CAUSE_TIME_OUT + ".");
throw e;
}
}
private static void checkAuthenticateOverKrb() throws IOException {
UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
if (loginUser == null) {
LOG.error("current user is " + currentUser + ", but loginUser is null.");
throw new IOException("current user is " + currentUser + ", but loginUser is null.");
}
if (!loginUser.equals(currentUser)) {
LOG.error("current user is " + currentUser + ", but loginUser is " + loginUser + ".");
throw new IOException("current user is " + currentUser + ", but loginUser is " + loginUser + ".");
}
if (!loginUser.hasKerberosCredentials()) {
LOG.error("current user is " + currentUser + " has no Kerberos Credentials.");
throw new IOException("current user is " + currentUser + " has no Kerberos Credentials.");
}
if (!UserGroupInformation.isLoginKeytabBased()) {
LOG.error("current user is " + currentUser + " is not Login Keytab Based.");
throw new IOException("current user is " + currentUser + " is not Login Keytab Based.");
}
}
private static boolean checkCurrentUserCorrect(String principal) throws IOException {
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
if (ugi == null) {
LOG.error("current user still null.");
throw new IOException("current user still null.");
}
String defaultRealm = null;
try {
defaultRealm = KerberosUtil.getDefaultRealm();
} catch (Exception e) {
LOG.warn("getDefaultRealm failed.");
throw new IOException(e);
}
if ((defaultRealm != null) && (defaultRealm.length() > 0)) {
StringBuilder realm = new StringBuilder();
StringBuilder principalWithRealm = new StringBuilder();
realm.append("@").append(defaultRealm);
if (!principal.endsWith(realm.toString())) {
principalWithRealm.append(principal).append(realm);
principal = principalWithRealm.toString();
}
}
return principal.equals(ugi.getUserName());
}
public static void loginWithUserKeytab(String uPrincipal) throws IOException {
String userPrincipal = uPrincipal;
String userKeytabPath = System.getProperty("user.dir") + File.separator + "user.keytab";
String krb5ConfPath = System.getProperty("user.dir") + File.separator + "krb5.conf";
String principalName = com.mrs.spark.security.hadoop.KerberosUtil.getKrb5DomainRealm();
LOG.warn("login using user super. principal name is:" + principalName);
String ZKServerPrincipal = "zookeeper/hadoop." + principalName;
String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client";
String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
Configuration hadoopConf = new Configuration();
try {
// setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userPrincipal, userKeytabPath);
setJaasFile(userPrincipal, userKeytabPath);
setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZKServerPrincipal);
login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf);
} catch (IOException e) {
LOG.warn("Login with keytab failed.");
throw new IOException(e);
}
}
/**
* copy from hbase zkutil 0.94&0.98 A JAAS configuration that defines the login modules that we want to use for
* login.
*/
private static class JaasConfiguration extends javax.security.auth.login.Configuration {
private static final Map<String, String> BASIC_JAAS_OPTIONS = new HashMap<String, String>();
static {
String jaasEnvVar = System.getenv("HBASE_JAAS_DEBUG");
if (jaasEnvVar != null && "true".equalsIgnoreCase(jaasEnvVar)) {
BASIC_JAAS_OPTIONS.put("debug", "true");
}
}
private static final Map<String, String> KEYTAB_KERBEROS_OPTIONS = new HashMap<String, String>();
static {
if (IS_IBM_JDK) {
KEYTAB_KERBEROS_OPTIONS.put("credsType", "both");
} else {
KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
KEYTAB_KERBEROS_OPTIONS.put("useTicketCache", "false");
KEYTAB_KERBEROS_OPTIONS.put("doNotPrompt", "true");
KEYTAB_KERBEROS_OPTIONS.put("storeKey", "true");
}
KEYTAB_KERBEROS_OPTIONS.putAll(BASIC_JAAS_OPTIONS);
}
private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN =
new AppConfigurationEntry(
KerberosUtil.getKrb5LoginModuleName(),
LoginModuleControlFlag.REQUIRED,
KEYTAB_KERBEROS_OPTIONS);
private static final AppConfigurationEntry[] KEYTAB_KERBEROS_CONF =
new AppConfigurationEntry[] {KEYTAB_KERBEROS_LOGIN};
private javax.security.auth.login.Configuration baseConfig;
private final String loginContextName;
private final boolean useTicketCache;
private final String keytabFile;
private final String principal;
public JaasConfiguration(String loginContextName, String principal, String keytabFile) throws IOException {
this(loginContextName, principal, keytabFile, keytabFile == null || keytabFile.length() == 0);
}
private JaasConfiguration(String loginContextName, String principal, String keytabFile, boolean useTicketCache)
throws IOException {
try {
this.baseConfig = javax.security.auth.login.Configuration.getConfiguration();
} catch (SecurityException e) {
this.baseConfig = null;
}
this.loginContextName = loginContextName;
this.useTicketCache = useTicketCache;
this.keytabFile = keytabFile;
this.principal = principal;
initKerberosOption();
LOG.info(
"JaasConfiguration loginContextName="
+ loginContextName
+ " principal="
+ principal
+ " useTicketCache="
+ useTicketCache
+ " keytabFile="
+ keytabFile);
}
private void initKerberosOption() throws IOException {
if (!useTicketCache) {
if (IS_IBM_JDK) {
KEYTAB_KERBEROS_OPTIONS.put("useKeytab", keytabFile);
} else {
KEYTAB_KERBEROS_OPTIONS.put("keyTab", keytabFile);
KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
KEYTAB_KERBEROS_OPTIONS.put("useTicketCache", useTicketCache ? "true" : "false");
}
}
KEYTAB_KERBEROS_OPTIONS.put("principal", principal);
}
public AppConfigurationEntry[] getAppConfigurationEntry(String appName) {
if (loginContextName.equals(appName)) {
return KEYTAB_KERBEROS_CONF;
}
if (baseConfig != null) return baseConfig.getAppConfigurationEntry(appName);
return (null);
}
}
}

复制代码

最后
谢谢各人 @500佰

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

spark-submit cluster模式业务开发 --解决配置文件读取不到、安全认证不 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

spark-submit cluster模式业务开发 --解决 配置文件读取不到、 安全认证不 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

spark-submit cluster模式业务开发 --解决配置文件读取不到、安全认证不 ...