07、FlinkSQL - Flink 1.10+通过Kerberos连接HiveCatalog

背景

因为要开发Flinksql，决定要使用HiveCatalog的支持，Flink当前最新版本是1.12.2，集群Hive的版本是1.1.0，而且需要用某个Linux用户进行代理。

在实际开发中，遇到两个问题：

1、 Hive1.1.0使用的不是jdbc，而是MetastoreClient，通过Thrift进行连接，而他不支持HADOOP_PROXY_USER；
2、 Kerberos认证需要什么配置文件，是否需要在代码里配置UGI？；

问题一：HADOOP_PROXY_USER支持

这个问题上一篇文章 已经给出解决方案。

具体请参考：hive-metastore（HIVE-15579）
Github代码CommitGithub链接

问题二：Kerberos认证需要什么配置文件，是否需要在代码里配置UGI？

经过测试，发现并不需要在代码中写任何 UserGroupInformation 和 doAs 相关的代码。

需要的配置文件如下：
hive-site.xml

<!--First created by Slankka-->
<configuration>
  <property>
    <name>hive.metastore.uris</name>
    <value>thrift://xxxxx.xxxxx.xxxxxx.com:xxxxx</value>
  </property>
  <property>
    <name>hive.metastore.client.socket.timeout</name>
    <value>300</value>
  </property>
  <!--property>
    <name>hive.metastore.execute.setugi</name>
    <value>slankka</value>
  </property-->
  <property>
    <name>hive.cluster.delegation.token.store.class</name>
    <value>org.apache.hadoop.hive.thrift.MemoryTokenStore</value>
  </property>
  <!--property>
    <name>hive.server2.enable.doAs</name>
    <value>true</value>
  </property-->
  <property>
    <name>hive.metastore.sasl.enabled</name>
    <value>true</value>
  </property>
  <!--property>
    <name>hive.server2.authentication</name>
    <value>kerberos</value>
  </property-->
  <property>
    <name>hive.metastore.kerberos.principal</name>
    <value>hive/_HOST@slankka.COM</value>
  </property>
  <property>
    <name>hive.server2.authentication.kerberos.principal</name>
    <value>hive/_HOST@slankka.COM</value>
  </property>
</configuration>

另外还需要一个文件:
core-site.xml

<configuration>
  <property>
    <name>hadoop.security.authentication</name>
    <value>kerberos</value>
  </property>
</configuration>

这样在程序启动的时候，只需要指定这两个配置文件即可。

另外不需要任何HADOOP_CONF_DIR或者HIVE_CONF_DIR。

以上内容就是最小化配置。

代码示例：

  public static void main(String[] args) {
        Catalog catalog = new HiveCatalog("slankka", "flink", args[0], args[1], "1.1.0");

        try {
//            List<String> strings = catalog.listDatabases();
//            for (String database : strings) {
//                System.out.println(database);
//            }
//            ObjectPath objectPath = new ObjectPath("flink", "objectName");
//            catalog.createFunction(objectPath, new CatalogFunctionImpl("className", FunctionLanguage.JAVA), false);
//            catalog.dropFunction(objectPath, false);
//            catalog.alterFunction(objectPath, new CatalogFunctionImpl("className", FunctionLanguage.JAVA), false);
//            CatalogFunction function = catalog.getFunction(objectPath);
//            catalog.listFunctions("flink");
//            catalog.createTable(objectPath, new CatalogTableImpl());
            catalog.open();

            EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
            TableEnvironment tableEnv = TableEnvironment.create(settings);
            String[] strings = tableEnv.listCatalogs();
            Arrays.stream(strings).forEach(System.out::println);
            boolean pfc = Arrays.asList(strings).contains("slankka");
            if (!pfc) {
                tableEnv.registerCatalog("slankka", catalog);
            }
            tableEnv.useCatalog("slankka");
            tableEnv.useDatabase("flink");
            tableEnv.executeSql("drop table if exists slankka.flink.WordCountSink");
            TableResult tableResult = tableEnv.executeSql("create table slankka.flink.WordCountSink (\n" +
                    "   word STRING,\n" +
                    "   len INT\n" +
                    ") WITH (\n" +
                    "   'connector' = 'jdbc',\n" +
                    "   'url' = 'jdbc:mysql://slankka.com:3306/rtc',\n" +
                    "   'table-name' = 'flink_sink_test',\n" +
                    "   'username' = 'root',\n" +
                    "   'password' = '1'\n" +
                    ")");
            tableResult.print();
            String[] tables = tableEnv.listTables();
            System.out.println("Tables: --->");
            Arrays.stream(tables).forEach(System.out::println);

        } finally {
            catalog.close();
        }
    }
}

执行结果如下：

21/03/15 15:23:08 INFO hive.HiveCatalog: Setting hive conf dir as /data/work/cloudservice/slankka/lib/
21/03/15 15:23:09 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
21/03/15 15:23:09 INFO hive.HiveCatalog: Created HiveCatalog 'slankka'
21/03/15 15:23:09 INFO hive.metastore: HADOOP_PROXY_USER is set. Using delegation token for HiveMetaStore connection.
21/03/15 15:23:09 INFO hive.metastore: Trying to connect to metastore with URI thrift://xxxxx.xxxxx.xxxxx.com:xxxx
21/03/15 15:23:09 INFO hive.metastore: Opened a connection to metastore, current connections: 1
21/03/15 15:23:09 INFO hive.metastore: Connected to metastore.
21/03/15 15:23:09 INFO hive.metastore: Closed a connection to metastore, current connections: 0
21/03/15 15:23:09 INFO hive.metastore: Trying to connect to metastore with URI thrift://xxxxx.xxxxx.xxxxx.com:xxxx
21/03/15 15:23:09 INFO hive.metastore: Opened a connection to metastore, current connections: 1
21/03/15 15:23:09 INFO hive.metastore: Connected to metastore.
21/03/15 15:23:09 INFO hive.HiveCatalog: Connected to Hive metastore
default_catalog
21/03/15 15:23:10 INFO catalog.CatalogManager: Set the current default catalog as [slankka] and the current default database as [flink].
+--------+
| result |
+--------+
|     OK |
+--------+
1 row in set
Tables: --->
wordcountsink