如何使用 Kerberos 在 hive UDF 中获得 HBase 连接？

Question

我想写一个UDF从HBase获取一些东西，我用它来设置token到hiveconf，但是我不能用hiveconf连接到HBase，它会抛出NullPointException

我尝试了很多方法，比如：https://www.programcreek.com/java-api-examples/index.php?api=org.apache.hadoop.hbase.security.token.TokenUtil

但它仍然抛出 NullPointException

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.Connection;

import org.apache.hadoop.hbase.client.ConnectionFactory;

import org.apache.hadoop.hbase.security.User;

import org.apache.hadoop.hbase.security.token.AuthenticationTokenIdentifier;

import org.apache.hadoop.hbase.security.token.TokenUtil;

import org.apache.hadoop.hive.conf.HiveConf;

import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext;

import org.apache.hadoop.hive.ql.hooks.HookContext;

import org.apache.hadoop.security.UserGroupInformation;

import org.apache.hadoop.security.token.Token;

public class HbaseTokenFetcherHook implements ExecuteWithHookContext{   

    private static final Log LOG = LogFactory.getLog(HbaseTokenFetcherHook.class);

    @Override

    public void run(HookContext hookContext) throws Exception {

          HiveConf hiveConf = hookContext.getConf();

          /* If required */

          hiveConf.set(“zookeeper.znode.parent”, "/hbase-secure");   

          try {               

               UserGroupInformation.setConfiguration(hiveConf);

               Connection tokenConnection = ConnectionFactory.createConnection(hiveConf);

               Token<AuthenticationTokenIdentifier> token = TokenUtil.obtainToken(tokenConnection, User.getCurrent());

               String urlString = token.encodeToUrlString();

               hiveConf.set(“HBASE_AUTH_TOKEN”, urlString);

          } catch (IOException | InterruptedException e) {

               LOG.error("Error while fetching token for hbase"

                         + e.getMessage(), e);

          }

     }

}

它抛出异常：

Token<AuthenticationTokenIdentifier> token = TokenUtil.obtainToken(tokenConnection, User.getCurrent());

错误信息： 引起：java.lang.NullPointException 在 org.apache.hadoop.hbase.zookeeper.ZookeeperWatcher.getMetaReplicaNodes(ZookeeperWatcher.java:497) 在 org.apache.hadoop.hbase.zookeeper.MetaTableLocator.blockUntilAvailable(MetaTableLocator.java:558)

Answer 1

试试Propolis第三方Hive UDF，它有各种HBase get函数来获取值或全家。 README 中提供了如何构建和使用的说明，以获取有关特定 UDF 类型 describe 和函数名称的更多信息。我使用 Kerberos 在 Hadoop 集群上对其进行了测试，它运行良好。

Answer 2

要获得使用 UserGroupInformation 初始化的配置对象（查看this），您可能需要提供更多信息。您可能认为从 hookContext.getConf 获得的 hiveConf 应该具有所有 Kerberos 所需的配置，但它可能没有。也许这段代码可以给你一个线索：

import javax.security.auth.login.AppConfigurationEntry;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

public String obtainHBASEDataWithKerberos(String key, String namespace, String tableName, String family, String qualifier) {
    try {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181, node2:2181, node3:2181");
        conf.set("hadoop.security.authentication","Kerberos");
        System.setProperty("javax.security.auth.useSubjectCredsOnly","false"); // https://stackoverflow.com/questions/33829017/gssexception-no-valid-credentials-provided-mechanism-level-failed-to-find-any
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conf.set("hbase.cluster.distributed","true");       
        conf.set("zookeeper.znode.parent","/hbase-secure");
        conf.set("hbase.security.authentication", "Kerberos");
        UserGroupInformation.setConfiguration(conf);
        UserGroupInformation.loginUserFromSubject(null);
        UserGroupInformation ugi=UserGroupInformation.getLoginUser();
        String kerberos_principal=ugi.getUserName();
        if (kerberos_principal!=null) {
            if (kerberos_principal.contains("@")) {
                String domain=kerberos_principal.split("@")[1];
                conf.set("hbase.master.kerberos.principal", "hbase/_HOST@"+domain);
                conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@"+domain);
            }
            // Create in-memory jaas file
            // Create HBASE entry options.
            HashMap<String, Object> hbase_options = new HashMap<String, Object>();
            hbase_options.put("doNotPrompt", "true");
            hbase_options.put("useTicketCache", "true");
            hbase_options.put("principal",kerberos_principal);
            hbase_options.put("storeKey","true");
            hbase_options.put("debug","true");

            // Create entries.
            final AppConfigurationEntry[] hbase_entries = {
                    new AppConfigurationEntry(
                            "com.sun.security.auth.module.Krb5LoginModule",
                            AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
                            hbase_options)
            };

            // Create configuration.
            javax.security.auth.login.Configuration jaasConfig = new javax.security.auth.login.Configuration() {
                @Override
                public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
                    if ("Client".equals(name)) {
                        return hbase_entries;
                    }
                    else return null;
                }
            };

            javax.security.auth.login.Configuration.setConfiguration(jaasConfig);
            UserGroupInformation.setConfiguration(conf);

            Connection conn = null;
            Table table = null;

            conn=CentralKerberosUGI.getHBaseConnection();
            table = conn.getTable(TableName.valueOf(namespace + ":" + tableName));
            Result result = table.get(new Get(key.getBytes()));
            byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            return Bytes.toString(value);
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}