华为云用户手册

接口使用建议 建议使用org.apache.hadoop.hbase.Cell作为KV数据对象，而不是org.apache.hadoop.hbase.KeyValue。 建议使用Connection connection = ConnectionFactory.createConnection(conf)来创建连接，废弃HTablePool。 建议使用org.apache.hadoop.hbase.mapreduce，不建议使用org.apache.hadoop.hbase.mapred。 建议通过构造出来的Connection对象的getAdmin()方法来获取HBase的客户端操作对象。

HBase接口类型介绍 由于HBase本身是由java语言开发出来的，且java语言具有简洁通用易懂的特性，推荐用户使用java语言进行HBase应用程序开发。 HBase采用的接口与Apache HBase保持一致。 HBase通过接口调用，可提供的功能如表1所示。 表1 HBase接口提供的功能 功能 说明 CRUD数据读写功能 增查改删。 高级特性 过滤器、二级索引、协处理器。 管理功能 表管理、集群管理。

操作场景 为了运行 FusionInsight MRS 产品Flink组件的SpringBoot接口样例代码，需要完成下面的操作。当前支持 GaussDB （DWS）样例工程。 该章节以在Linux环境下开发GaussDB（DWS） SpringBoot方式连接Flink服务的应用程序为例。 执行GaussDB（DWS）样例需提前登录GaussDB（DWS）所在节点创建用于接受数据的空表“test_lzh1”，创建命令如下： create table test_lzh1 (id integer not null);

使用Python提交Flink普通作业 获取样例工程“flink-examples/pyflink-example/pyflink-kafka”中的“pyflink-kafka.py”和“insertData2kafka.sql”。 参考准备本地应用开发环境将准备好的Python虚拟环境打包，获取“venv.zip”文件。 zip -q -r venv.zip venv/ 以root用户登录主管理节点，将1和2获取的“venv.zip”、“pyflink-kafka.py”和“insertData2kafka.sql”文件上传至客户端环境。 per-job模式：将上述文件上传到“客户端安装目录/Flink/flink”。 yarn-application模式：将上述文件和“flink-connector-kafka-实际版本号.jar”包上传到“客户端安装目录/Flink/flink/yarnship”。 修改“pyflink-kafka.py”中的“specific_jars”路径。 per-job模式：修改为SQL文件的实际路径。如：file:///客户端安装目录/Flink/flink/lib/flink-connector-kafka-实际版本号.jar yarn-application模式：修改为：file://"+os.getcwd()+"/../../../../yarnship/flink-connector-kafka-实际版本号.jar 修改“pyflink-kafka.py”中的“file_path”路径。 per-job模式：修改为sql文件的实际路径。如：客户端安装目录/Flink/flink/insertData2kafka.sql yarn-application模式：修改为os.getcwd() + "/../../../../yarnship/insertData2kafka.sql" 执行以下命令指定运行环境。 export PYFLINK_CLIENT_EXECUTABLE=venv.zip/venv/bin/python3 执行以下命令运行程序。 per-job模式： ./bin/flink run --detached -t yarn-per-job -Dyarn.application.name=py_kafka -pyarch venv.zip -pyexec venv.zip/venv/bin/python3 -py pyflink-kafka.py 运行结果： yarn-application模式 ./bin/flink run-application --detached -t yarn-application -Dyarn.application.name=py_kafka -Dyarn.ship-files=/opt/client/Flink/flink/yarnship/ -pyarch yarnship/venv.zip -pyexec venv.zip/venv/bin/python3 -pyclientexec venv.zip/venv/bin/python3 -pyfs yarnship -pym pyflink-kafka 运行结果： 父主题： PyFlink样例程序

回答 建议将"blob.storage.directory"配置选项设置成“/tmp”或者“/opt/huawei/Bigdata/tmp”。 当用户将"blob.storage.directory"配置选项设置成自定义目录时，需要手动赋予用户该目录的owner权限。以下以FusionInsight的admin用户为例。 修改Flink客户端配置文件conf/flink-conf.yaml，配置blob.storage.directory: /home/testdir/testdirdir/xxx。 创建目录/home/testdir（创建一层目录即可），设置该目录为admin用户所属。 /home/testdir/下的testdirdir/xxx目录在启动Flink集群时会在每个节点下自动创建。 进入客户端路径，执行命令./bin/yarn-session.sh -jm 2048 -tm 3072，可以看到yarn-session正常启动并且成功创建目录。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的testDelete方法中。 public void testDelete() { LOG .info("Entering testDelete."); byte[] rowKey = Bytes.toBytes("012005000201"); Table table = null; try { // Instantiate an HTable object. table = conn.getTable(tableName); // Instantiate an Delete object. Delete delete = new Delete(rowKey); // Submit a delete request. table.delete(delete); LOG.info("Delete table successfully."); } catch (IOException e) { LOG.error("Delete table failed " ,e); } finally { if (table != null) { try { // Close the HTable object. table.close(); } catch (IOException e) { LOG.error("Close table failed " ,e); } } } LOG.info("Exiting testDelete."); } 如果被删除的cell所在的列族上设置了二级索引，也会同步删除索引数据。

前提条件 已安装客户端时： 已安装HBase客户端。 当客户端所在主机不是集群中的节点时，需要在客户端所在节点的hosts文件中设置主机名和IP地址映射。主机名和IP地址请保持一一对应。 未安装HBase客户端时： Linux环境已安装JDK，版本号需要和IntelliJ IDEA导出Jar包使用的JDK版本一致。 当Linux环境所在主机不是集群中的节点时，需要在节点的hosts文件中设置主机名和IP地址映射。主机名和IP地址请保持一一对应。

场景说明 在安全集群环境下，各个组件之间的相互通信不能够简单地互通，而需要在通信之前进行相互认证，以确保通信的安全性。HBase应用开发需要进行ZooKeeper和Kerberos安全认证。用于ZooKeeper认证的文件为“jaas.conf”，用于Kerberos安全认证文件为keytab文件和krb5.conf文件。具体使用方法在样例代码的“README.md”中会有详细说明。 jaas.conf文件请参考获取MRS应用开发样例工程，进入“src/hbase-examples/hbase-zk-example/src/main/resources/”路径下获取。 keytab和krb5.conf文件获取方法请参考准备MRS应用开发用户。 安全认证主要采用代码认证方式。支持Oracle JAVA平台和IBM JAVA平台。 以下代码在“com.huawei.bigdata.hbase.examples”包的“TestMain”类中。

场景说明 当不同的多个Manager系统下安全模式的集群需要互相访问对方的资源时，管理员可以设置互信的系统，使外部系统的用户可以在本系统中使用。每个系统用户安全使用的范围定义为“域”，不同的Manager系统需要定义唯一的 域名 。跨Manager访问实际上就是用户跨域使用。集群配置互信具体操作步骤请参考集群互信管理章节。 多集群互信场景下，以符合跨域访问的用户身份，使用从其中一个manager系统中获取到的用于Kerberos安全认证的keytab文件和principal文件，以及多个Manager系统各自的客户端配置文件，可实现一次认证登录后访问调用多集群的HBase服务。 以下代码在hbase-example样例工程的“com.huawei.bigdata.hbase.examples”包的“TestMultipleLogin”类中。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“PhoenixSample”类的testCreateTable方法中。 /** * Create Table */ public void testCreateTable() { LOG.info("Entering testCreateTable."); String URL = "jdbc:phoenix:" + conf.get("hbase.zookeeper.quorum"); // Create table String createTableSQL = "CREATE TABLE IF NOT EXISTS TEST (id integer not null primary key, name varchar, " + "account char(6), birth date)"; try (Connection conn = DriverManager.getConnection(url, props); Statement stat = conn.createStatement()) { // Execute Create SQL stat.executeUpdate(createTableSQL); LOG.info("Create table successfully."); } catch (Exception e) { LOG.error("Create table failed.", e); } LOG.info("Exiting testCreateTable."); } /** * Drop Table */ public void testDrop() { LOG.info("Entering testDrop."); String URL = "jdbc:phoenix:" + conf.get("hbase.zookeeper.quorum"); // Delete table String dropTableSQL = "DROP TABLE TEST"; try (Connection conn = DriverManager.getConnection(url, props); Statement stat = conn.createStatement()) { stat.executeUpdate(dropTableSQL); LOG.info("Drop successfully."); } catch (Exception e) { LOG.error("Drop failed.", e); } LOG.info("Exiting testDrop."); }

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“PhoenixSample”类的testSelect方法中。 /** * Select Data */ public void testSelect() { LOG.info("Entering testSelect."); String URL = "jdbc:phoenix:" + conf.get("hbase.zookeeper.quorum"); // Query String querySQL = "SELECT * FROM TEST WHERE id = ?"; Connection conn = null; PreparedStatement preStat = null; Statement stat = null; ResultSet result = null; try { // Create Connection conn = DriverManager.getConnection(url, props); // Create Statement stat = conn.createStatement(); // Create PrepareStatement preStat = conn.prepareStatement(querySQL); // Execute query preStat.setInt(1, 1); result = preStat.executeQuery(); // Get result while (result.next()) { int id = result.getInt("id"); String name = result.getString(1); System.out.println("id: " + id); System.out.println("name: " + name); } LOG.info("Select successfully."); } catch (Exception e) { LOG.error("Select failed.", e); } finally { if (null != result) { try { result.close(); } catch (Exception e2) { LOG.error("Result close failed.", e2); } } if (null != stat) { try { stat.close(); } catch (Exception e2) { LOG.error("Stat close failed.", e2); } } if (null != conn) { try { conn.close(); } catch (Exception e2) { LOG.error("Connection close failed.", e2); } } } LOG.info("Exiting testSelect."); }

场景说明 假定用户开发一个应用程序，用于管理企业中的使用A业务的用户信息，如表1所示，A业务操作流程如下： 创建用户信息表。 在用户信息中新增用户的学历、职称等信息。 根据用户编号查询用户姓名和地址。 根据用户姓名进行查询。 查询年龄段在[20-29]之间的用户信息。 数据统计，统计用户信息表的人员数、年龄最大值、年龄最小值、平均年龄。 用户销户，删除用户信息表中该用户的数据。 A业务结束后，删除用户信息表。 表1 用户信息 编号 姓名 性别 年龄 地址 12005000201 张三 男 19 广东省深圳市 12005000202 李婉婷 女 23 河北省石家庄市 12005000203 王明 男 26 浙江省宁波市 12005000204 李刚 男 18 湖北省襄阳市 12005000205 赵恩如 女 21 江西省上饶市 12005000206 陈龙 男 32 湖南省株洲市 12005000207 周微 女 29 河南省南阳市 12005000208 杨艺文 女 30 重庆市开县 12005000209 徐兵 男 26 陕西省渭南市 12005000210 肖凯 男 25 辽宁省大连市

开发思路 根据上述的业务场景进行功能分解，需要开发的功能点如表2所示。 表2 在HBase中开发的功能 序号 步骤 代码实现 1 根据表1中的信息创建表。 请参见创建HBase表。 2 导入用户数据。 请参见向HBase表中插入数据。 3 增加“教育信息”列族，在用户信息中新增用户的学历、职称等信息。 请参见修改HBase表。 4 根据用户编号查询用户姓名和地址。 请参见使用Get API读取HBase表数据。 5 根据用户姓名进行查询。 请参见使用Filter过滤器读取HBase表数据。 6 为提升查询性能，创建二级索引或者删除二级索引。 请参见创建HBase表二级索引和基于二级索引查询HBase表数据。 7 用户销户，删除用户信息表中该用户的数据。 请参见删除HBase表数据。 8 A业务结束后，删除用户信息表。 请参见删除HBase表。

问题 Flink任务配置State Backend为RocksDB时，运行报如下错误： Caused by: java.lang.UnsatisfiedLinkError: /srv/BigData/hadoop/data1/nm/usercache/***/appcache/application_****/rocksdb-lib-****/librocksdbjni-linux64.so: /lib64/libpthread.so.0: version `GLIBC_2.12` not found (required by /srv/BigData/hadoop/***/librocksdbjni-linux64.so)at java.lang.ClassLoader$NativeLibrary.load(Native Method) at java.lang.ClassLoader.loadLibrary0(ClassLoader.java:1965) at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1890) at java.lang.Runtime.load0(Runtime.java:795) at java.lang.System.load(System.java:1062) at org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:78)at org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:56)at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.ensureRocksDBIsLoaded(RocksDBStateBackend.java:734)... 11 more

注意事项 如果yarn-session.sh使用-z配置特定的zookeeper的namespace，则在使用flink run时必须使用-yid指出applicationID，使用-yz指出zookeeper的namespace，前后namespace保持一致。 举例： bin/yarn-session.sh -z YARN101 bin/flink run -yid application_****_**** -yz YARN101 examples/streaming/WindowJoin.jar

回答 首先查看ZooKeeper中/flink_base的目录权限是否为：'world,'anyone: cdrwa；如果不是，请修改/flink_base的目录权限为：'world,'anyone: cdrwa，然后继续根据步骤二排查；如果是，请根据步骤二排查。 由于在Flink配置文件中“high-availability.zookeeper.client.acl”默认为“creator”，即谁创建谁有权限，由于原有用户已经使用ZooKeeper上的/flink_base/flink目录，导致新创建的用户访问不了ZooKeeper上的/flink_base/flink目录。 新用户可以通过以下操作来解决问题。 查看客户端的配置文件“conf/flink-conf.yaml”。 修改配置项“high-availability.zookeeper.path.root”对应的ZooKeeper目录，例如：/flink2。 重新提交任务。

开发思路 写HBase： 通过参数指定“hbase-site.xml”文件的父目录，Flink Sink可以获取到HBase的Connection。 通过Connection判断表是否存在，如果不存在则创建表。 将接收到的数据转化成Put对象，写到HBase。 读HBase： 通过参数指定“hbase-site.xml”文件的父目录，Flink Source可以获取到HBase的Connection。 通过Connection判断表是否存在，如果不存在则作业失败，需要通过HBase Shell创建表或上游作业创建表。 读取HBase中的数据，将Result数据转化成Row对象发送给下游算子。

代码样例 以下为代码示例： hbase.root.logger=INFO,console,RFA //hbase客户端日志输出配置，console：输出到控制台；RFA：输出到日志文件hbase.security.logger=DEBUG,console,RFAS //hbase客户端安全相关的日志输出配置，console：输出到控制台；RFAS：输出到日志文件hbase.log.dir=/var/log/Bigdata/hbase/client/ //日志路径，根据实际路径修改，但目录要有写入权限hbase.log.file=hbase-client.log //日志文件名hbase.log.level=INFO //日志级别，如果需要更详细的日志定位问题，需要修改为DEBUG，修改完需要重启进程才能生效hbase.log.maxbackupindex=20 //最多保存的日志文件数目# Security audit appenderhbase.security.log.file=hbase-client-audit.log //审计日志文件命令

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的dropTable方法中。 public void dropTable() { LOG.info("Entering dropTable."); Admin admin = null; try { admin = conn.getAdmin(); if (admin.tableExists(tableName)) { // Disable the table before deleting it. admin.disableTable(tableName); // Delete table. admin.deleteTable(tableName);//注[1] } LOG.info("Drop table successfully."); } catch (IOException e) { LOG.error("Drop table failed " ,e); } finally { if (admin != null) { try { // Close the Admin object. admin.close(); } catch (IOException e) { LOG.error("Close admin failed " ,e); } } } LOG.info("Exiting dropTable."); }

操作场景 HBase把Thrift结合起来可以向外部应用提供HBase服务。在HBase服务安装时可选部署ThriftServer实例，ThriftServer系统可访问HBase的用户，拥有HBase所有NameSpace和表的读、写、执行、创建和管理的权限。访问ThriftServer服务同样需要进行Kerberos认证。HBase实现了两套Thrift Server服务，此处“hbase-thrift-example”为ThriftServer实例服务的调用实现。

样例代码 代码认证 以下代码在“hbase-thrift-example”样例工程的“com.huawei.bigdata.hbase.examples”包的“TestMain”类中。 private static void init() throws IOException { // Default load from conf directory conf = HBaseConfiguration.create(); String userdir = TestMain.class.getClassLoader().getResource("conf").getPath() + File.separator;[1] //In Linux environment //String userdir = System.getProperty("user.dir") + File.separator + "conf" + File.separator; conf.addResource(new Path(userdir + "core-site.xml"), false); conf.addResource(new Path(userdir + "hdfs-site.xml"), false); conf.addResource(new Path(userdir + "hbase-site.xml"), false);} [1]userdir获取的是编译后资源路径下conf目录的路径。初始化配置用到的core-site.xml、hdfs-site.xml、hbase-site.xml文件和用于安全认证的用户凭证文件，需要放置到"src/main/resources/conf"的目录下。 安全登录 请根据实际情况，修改“userName”为实际用户名，例如“developuser”。 private static void login() throws IOException { if (User.isHBaseSecurityEnabled(conf)) { userName = " developuser "; //In Windows environment String userdir = TestMain.class.getClassLoader().getResource("conf").getPath() + File.separator; //In Linux environment //String userdir = System.getProperty("user.dir") + File.separator + "conf" + File.separator; userKeytabFile = userdir + "user.keytab"; krb5File = userdir + "krb5.conf"; /* * if need to connect zk, please provide jaas info about zk. of course, * you can do it as below: * System.setProperty("java.security.auth.login.config", confDirPath + * "jaas.conf"); but the demo can help you more : Note: if this process * will connect more than one zk cluster, the demo may be not proper. you * can contact us for more help */ LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userName, userKeytabFile); LoginUtil.login(userName, userKeytabFile, krb5File, conf); }} 连接ThriftServer实例 try { test = new ThriftSample(); test.test("10.120.16.170", THRIFT_PORT, conf);[2] } catch (TException | IOException e) { LOG.error("Test thrift error", e); } [2]test.test()传入参数为待访问的ThriftServer实例所在节点ip地址，需根据实际运行集群情况进行修改，且该节点ip需要配置到运行样例代码的本机hosts文件中。 “THRIFT_PORT”为ThriftServer实例的配置参数"hbase.regionserver.thrift.port"对应的值。

问题 Flink内核升级到1.3.0之后，当Kafka调用带有非static的KafkaPartitioner类对象为参数的FlinkKafkaProducer010去构造函数时，运行时会报错。 报错内容如下： org.apache.flink.api.common.InvalidProgramException: The implementation of the FlinkKafkaPartitioner is not serializable. The object probably contains or references non serializable fields.

数据规划 准备用户认证文件：登录Manager下载用户凭证，获取“user.keytab”和“krb5.conf”文件。 准备待创建的租户信息，如“tenantId”为“92”，“tenantName”为“test92”，“remark”为“test tenant remark1”。 如果是在Windows运行本样例程序，需将所有FlinkServer所在节点的主机名和IP地址添加到“C:\Windows\System32\drivers\etc\hosts”中。

提供Join能力 表12 提供Join能力的相关接口 API 说明 def join[T2](otherStream: DataStream[T2]): JoinedStreams[T, T2] 通过给定的key在一个窗口范围内join两条数据流。 join操作的key值通过where和eaualTo方法进行指定，代表两条流过滤出包含等值条件的数据。 def coGroup[T2](otherStream: DataStream[T2]): CoGroupedStreams[T, T2] 通过给定的key在一个窗口范围内co-group两条数据流。 coGroup操作的key值通过where和eaualTo方法进行指定，代表两条流通过该等值条件进行分区处理。

Flink常用接口 Flink主要使用到如下这几个类： StreamExecutionEnvironment：是Flink流处理的基础，提供了程序的执行环境。 DataStream：Flink用特别的类DataStream来表示程序中的流式数据。用户可以认为它们是含有重复数据的不可修改的集合(collection)，DataStream中元素的数量是无限的。 KeyedStream：DataStream通过keyBy分组操作生成流，数据经过对设置的key值进行分组。 WindowedStream：KeyedStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 AllWindowedStream：DataStream通过window窗口函数生成的流，设置窗口类型并且定义窗口触发条件，然后在窗口数据上进行一些操作。 ConnectedStreams：将两条DataStream流连接起来并且保持原有流数据的类型，然后进行map或者flatMap操作。 JoinedStreams：在窗口上对数据进行等值join操作，join操作是coGroup操作的一种特殊场景。 CoGroupedStreams：在窗口上对数据进行coGroup操作，可以实现流的各种join类型。 图1 Flink Stream的各种流类型转换

提供设置eventtime属性的能力 表6 提供设置eventtime属性的能力的相关接口 API 说明 def assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T]): DataStream[T] 为了能让event time窗口可以正常触发窗口计算操作，需要从记录中提取时间戳。 def assignTimestampsAndWatermarks(assigner: AssignerWithPunctuatedWatermarks[T]): DataStream[T]

提供分流能力 表8 提供分流能力的相关接口 API 说明 def split(selector: OutputSelector[T]): SplitStream[T] 传入OutputSelector，重写select方法确定分流的依据(即打标记)，构建SplitStream流。即对每个元素做一个字符串的标记，作为选择的依据，打好标记之后就可以通过标记选出并新建某个标记的流。 def select(outputNames: String*): DataStream[T] 从一个SplitStream中选出一个或多个流。 outputNames指的是使用split方法对每个元素做的字符串标记的序列。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的testSingleColumnValueFilter方法中。 public void testSingleColumnValueFilter() { LOG.info("Entering testSingleColumnValueFilter."); Table table = null; ResultScanner rScanner = null; try { table = conn.getTable(tableName); Scan scan = new Scan(); scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name")); // Set the filter criteria. SingleColumnValueFilter filter = new SingleColumnValueFilter( Bytes.toBytes("info"), Bytes.toBytes("name"), CompareOperator.EQUAL, Bytes.toBytes("Xu Bing")); scan.setFilter(filter); // Submit a scan request. rScanner = table.getScanner(scan); // Print query results. for (Result r = rScanner.next(); r != null; r = rScanner.next()) { for (Cell cell : r.rawCells()) { LOG.info("{}:{},{},{}", Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneFamily(cell)), Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell))); } } LOG.info("Single column value filter successfully."); } catch (IOException e) { LOG.error("Single column value filter failed " ,e); } finally { if (rScanner != null) { // Close the scanner object. rScanner.close(); } if (table != null) { try { // Close the HTable object. table.close(); } catch (IOException e) { LOG.error("Close table failed " ,e); } } } LOG.info("Exiting testSingleColumnValueFilter."); }

注意事项 注[1]：创建联合索引。 HBase支持在多个字段上创建二级索引，例如在列name和age上。 HIndexSpecification iSpecUnite = new HIndexSpecification(indexName); iSpecUnite.addIndexColumn(new HColumnDescriptor("info"), "name", ValueType.String); iSpecUnite.addIndexColumn(new HColumnDescriptor("info"), "age", ValueType.String);

代码样例 以租户用户为“test92”，租户ID为“92”，获取具有FlinkServer管理员权限的用户名为“flinkserveradmin”的代理访问API为例，以下代码为完整示例。 public class TestCreateTenants { public static void main(String[] args) { ParameterTool paraTool = ParameterTool.fromArgs(args); final String hostName = paraTool.get("hostName"); // 修改hosts文件，使用主机名 final String keytab = paraTool.get("keytab"); // user.keytab路径 final String krb5 = paraTool.get("krb5"); // krb5.conf路径 final String principal = paraTool.get("principal"); // 认证用户 System.setProperty("java.security.krb5.conf", krb5); String url = "https://"+hostName+":28943/flink/v1/tenants"; String jsonstr = "{" + "\n\t \"tenantId\":\"92\"," + "\n\t \"tenantName\":\"test92\"," + "\n\t \"remark\":\"test tenant remark1\"," + "\n\t \"updateUser\":\"test_updateUser1\"," + "\n\t \"createUser\":\"test_createUser1\"" + "\n}"; try { LoginClient.getInstance().setConfigure(url, principal, keytab, ""); LoginClient.getInstance().login(); // 先使用flinkserver管理员用户登录 String proxyUrl = "https://"+hostName+":28943/flink/v1/proxyUserLogin"; // 调用代理用户接口，获取普通用户token String result = HttpClientUtil.doPost(proxyUrl, "{\n" + "\t\"realUser\": \"flinkserveradmin\"\n" + "}", "utf-8", true); Gson gson = new Gson(); JsonObject jsonObject = gson.fromJson(result, JsonObject.class); String token = jsonObject.get("result").toString(); token = "hadoop_auth=" + token; System.out.println(HttpClientUtil.doPost(url, jsonstr, "utf-8", true , token)); } catch (Exception e) { System.out.println(e); } }}