kerberos

spark

在sparksubmit中添加如下参数

--principal user@xxx.COM \
--keytab /home/user.keytab \

flink

在conf/flink-conf.yaml 中修改

security.kerberos.login.keytab: /home/user.keytab
security.kerberos.login.principal: user
security.kerberos.login.use-ticket-cache: true

hbase

需要在sparksubmit中加入

--driver-class-path /etc/hbase/conf \

flink程序中添加hbase-site.xml

Scala-无配置文件

//kerberos认证
System.setProperty("java.security.krb5.conf", kerberosConfPath)
def getHBaseConf() = {
  //hbase
  val hBaseConf = HBaseConfiguration.create()
  //conf.set("fs.permissions.umask-mode", "000")
  //hBaseConf.addResource(new Path("/etc/hbase/conf/hbase-site.xml"))
  hBaseConf.set("keytab.file", keytabPath)
  //这里需要设置的principal和下面user中设置principal不一样
  //hbase配置文件中的hbase.master.kerberos.principal
  hBaseConf.set("kerberos.principal", hBasePrincipal)
  hBaseConf.set("hbase.master.kerberos.principal", hBasePrincipal)
  hBaseConf.set("hbase.regionserver.kerberos.principal", hBasePrincipal)
  hBaseConf.set("hadoop.security.authentication", "kerberos")
  hBaseConf.set("hbase.security.authentication", "kerberos")
  hBaseConf.set("hbase.zookeeper.quorum", quorum)
  hBaseConf.set("hbase.zookeeper.property.clientPort", clientPort)
  //设置hfile最大个数
  hBaseConf.set("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", "3200")
  //设置hfile的大小
  hBaseConf.set("hbase.hregion.max.filesize", "10737418240")
  //kerberos认证
  UserGroupInformation.setConfiguration(hBaseConf)
  UserGroupInformation.loginUserFromKeytab(principal, keytabPath)
  //UserGroupInformation.getLoginUser.checkTGTAndReloginFromKeytab()
  hBaseConf
}

Java-有配置文件

public void setConnection() {
  if (connection == null || connection.isClosed() || connection.isAborted()) {
    System.setProperty("java.security.krb5.conf", kerberosConfPath);
    Configuration conf = HBaseConfiguration.create();
    conf.addResource(new Path("/etc/hbase/conf/hbase-site.xml"));
    UserGroupInformation.setConfiguration(conf);
    try {
      UserGroupInformation.loginUserFromKeytab(principal, keytabPath)
      connection = ConnectionFactory.createConnection(conf);
      log.info("获取Hbase连接:"+connection.toString());
    } catch (Exception e) {
      log.error("hbase 连接获取失败");
      e.printStackTrace();
    }
  }
}