华为云用户手册

通过HSBroker的用户名密码认证实现查询HetuEngine SQL任务 本章节适用于 MRS 3.3.0及以后版本。 通过HSBroker方式连接到HetuEngine，组装对应的SQL发送到HetuEngine执行，完成对Hive数据源的增删改查操作。 import jaydebeapi driver = "io.XXX.jdbc.XXXDriver" # need to change the value based on the cluster informationurl = "jdbc:XXX://192.168.37.61:29861,192.168.37.62:29861/hive/default?serviceDiscoveryMode=hsbroker"user = "YourUserName"tenant = "YourTenant"jdbc_location = "Your file path of the jdbc jar" sql = "show catalogs" if __name__ == '__main__': conn = jaydebeapi.connect(driver, url, {"user": user, "SSL": "false", "tenant": tenant}, [jdbc_location]) curs = conn.cursor() curs.execute(sql) result = curs.fetchall() print(result) curs.close() conn.close() 上述代码中各参数说明如下表所示： 表1 参数及参数说明 参数名称 参数说明 url jdbc:XXX://HSBroker1_IP:HSBroker1_Port,HSBroker2_IP:HSBroker2_Port,HSBroker3_IP:HSBroker3_Port/catalog/schema?serviceDiscoveryMode=hsbroker 说明： XXX：驱动名，请以实际样例代码中的内容为准。 catalog、schema分别是JDBC客户端要连接的catalog和schema名称。 HSBroker_IP:HSBroker_Port是HSBroker的URL，多个URL以逗号隔开。例如：“192.168.81.37:2181,192.168.195.232:2181,192.168.169.84:2181” user 访问HetuServer的用户名，即在集群中创建的“人机”用户的用户名。 tenant 指定访问HetuEngine计算实例的租户资源队列。 jdbc_location 导入并配置HetuEngine Python3样例工程中获取的hetu-jdbc-XXX.jar包的完整路径。 Windows系统路径示例："D:\\hetu-examples-python3\\hetu-jdbc-XXX.jar" Linux系统路径示例："/opt/hetu-examples-python3/hetu-jdbc-XXX.jar" 父主题： HetuEngine样例程序（Python3）

操作步骤 在Windows环境下的Intellij IDEA开发环境中，单击IDEA右侧Maven导入依赖。 图1 导入依赖 （可选）如果对接的集群开启了ZooKeeper的SSL认证通信功能，则需要添加JVM配置参数，如下所示： -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty -Dzookeeper.client.secure=true 在“JDBCExampleZK.java”文件下单击右键，在弹出菜单单击“Run 'JDBCExampleZK.main()' ”。 图2 运行程序 在IDEA的console中可以看到输出结果。 图3 输出结果

Hive介绍 Hive是一个开源的，建立在Hadoop上的 数据仓库 框架，提供类似SQL的HQL语言操作结构化数据，其基本原理是将HQL语言自动转换成MapReduce任务或Spark任务，从而完成对Hadoop集群中存储的海量数据进行查询和分析。 Hive主要特点如下： 通过HQL语言非常容易的完成数据提取、转换和加载（ETL）。 通过HQL完成海量结构化数据分析。 灵活的数据存储格式，支持JSON、 CS V、TEXTFILE、RCFILE、ORCFILE、SEQUENCEFILE等存储格式，并支持自定义扩展。 多种客户端连接方式，支持JDBC接口。 Hive主要应用于海量数据的离线分析（如 日志分析 ，集群状态分析）、大规模的数据挖掘（用户行为分析，兴趣分区，区域展示）等场景下。 为保证Hive服务的高可用性、用户数据的安全及访问服务的可控制，在开源社区的Hive-3.1.0版本基础上，Hive新增如下特性： 数据文件加密机制。 开源社区的Hive特性，请参见https://cwiki.apache.org/confluence/display/hive/designdocs。

准备开发环境 在进行应用开发时，要准备的开发和运行环境如表1所示。 表1 开发环境 准备项 说明 操作系统 开发环境：Windows系统，支持Windows7以上版本。 运行环境：Windows或Linux系统。 如需在本地调测程序，运行环境需要和集群业务平面网络互通。 安装JDK 开发和运行环境的基本配置，版本要求如下： 服务端和客户端仅支持自带的OpenJDK，版本为1.8.0_272，不允许替换。 对于客户应用需引用SDK类的Jar包运行在客户应用进程中的。 X86客户端：Oracle JDK：支持1.8版本；IBM JDK：支持1.8.5.11版本。 TaiShan客户端：OpenJDK：支持1.8.0_272版本。 说明： 基于安全考虑，服务端只支持TLS V1.2及以上的加密协议。 IBM JDK默认只支持TLS V1.0，若使用IBM JDK，请配置启动参数“com.ibm.jsse2.overrideDefaultTLS”为“true”，设置后可以同时支持TLS V1.0/V1.1/V1.2，详情参见https://www.ibm.com/support/knowledgecenter/zh/SSYKE2_8.0.0/com.ibm.java.security.component.80.doc/security-component/jsse2Docs/matchsslcontext_tls.html#matchsslcontext_tls。 安装和配置IntelliJ IDEA 开发环境的基本配置，建议使用2019.1或其他兼容版本。 说明： 若使用IBM JDK，请确保IntelliJ IDEA中的JDK配置为IBM JDK。 若使用Oracle JDK，请确保IntelliJ IDEA中的JDK配置为Oracle JDK。 若使用Open JDK，请确保IntelliJ IDEA中的JDK配置为Open JDK。 不同的IntelliJ IDEA不要使用相同的workspace和相同路径下的示例工程。 安装Maven 开发环境的基本配置。用于项目管理，贯穿软件开发生命周期。 7-zip 用于解压“*.zip”和“*.rar”文件。 支持7-Zip 16.04版本。

样例代码 -- 从本地文件系统/opt/hive_examples_data/目录下将employee_info.txt加载进employees_info表中.---- 用新数据覆盖原有数据 LOAD DATA LOCAL INPATH '/opt/hive_examples_data/employee_info.txt' OVERWRITE INTO TABLE employees_info; ---- 保留原有数据，将新数据追加到表中LOAD DATA LOCAL INPATH '/opt/hive_examples_data/employee_info.txt' INTO TABLE employees_info; -- 从HDFS上/user/hive_examples_data/employee_info.txt加载进employees_info表中. ---- 用新数据覆盖原有数据LOAD DATA INPATH '/user/hive_examples_data/employee_info.txt' OVERWRITE INTO TABLE employees_info; ---- 保留原有数据，将新数据追加到表中LOAD DATA INPATH '/user/hive_examples_data/employee_info.txt' INTO TABLE employees_info; 加载数据的实质是将数据复制到HDFS上指定表的目录下。

提供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方法进行指定，代表两条流通过该等值条件进行分区处理。

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

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]

代码样例 以下代码片段是登录，创建Connection并创建表的示例，在com.huawei.bigdata.hbase.examples包的“HBaseSample”类的HBaseSample方法中。 private TableName tableName = null; private Connection conn = null; public HBaseSample(Configuration conf) throws IOException { this.tableName = TableName.valueOf("hbase_sample_table"); this.conn = ConnectionFactory.createConnection(conf);}

代码样例 以下代码片段在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包的“HBaseSample”类的testGet方法中。 public void testGet() { LOG.info("Entering testGet."); // Specify the column family name. byte[] familyName = Bytes.toBytes("info"); // Specify the column name. byte[][] qualifier = { Bytes.toBytes("name"), Bytes.toBytes("address") }; // Specify RowKey. byte[] rowKey = Bytes.toBytes("012005000201"); Table table = null; try { // Create the Table instance. table = conn.getTable(tableName); // Instantiate a Get object. Get get = new Get(rowKey); // Set the column family name and column name. get.addColumn(familyName, qualifier[0]); get.addColumn(familyName, qualifier[1]); // Submit a get request. Result result = table.get(get); // Print query results. for (Cell cell : result.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("Get data successfully."); } catch (IOException e) { LOG.error("Get data 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 testGet."); }

功能介绍 HBase通过ConnectionFactory.createConnection(configuration)方法创建Connection对象。传递的参数为上一步创建的Configuration。 Connection封装了底层与各实际服务器的连接以及与ZooKeeper的连接。Connection通过ConnectionFactory类实例化。创建Connection是重量级操作，Connection是线程安全的，因此，多个客户端线程可以共享一个Connection。 典型的用法，一个客户端程序共享一个单独的Connection，每一个线程获取自己的Admin或Table实例，然后调用Admin对象或Table对象提供的操作接口。不建议缓存或者池化Table、Admin。Connection的生命周期由调用者维护，调用者通过调用close()，释放资源。

代码样例 以租户用户为“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); } }}

HDFS应用开发流程介绍 开发流程中各阶段的说明如图1和表1所示。 图1 HDFS应用程序开发流程 表1 HDFS应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解HDFS的基本概念。 HDFS应用开发简介 准备开发和运行环境 使用IntelliJ IDEA工具，请根据指导完成开发环境配置。 HDFS的运行环境即HDFS客户端，请根据指导完成客户端的安装和配置。 准备HDFS应用开发和运行环境 准备工程 HDFS提供了不同场景下的样例程序，可以导入样例工程进行程序学习。 导入并配置HDFS样例工程 根据场景开发工程 提供样例工程，帮助用户快速了解HDFS各部件的编程接口。 开发HDFS应用 编译并运行程序 指导用户将开发好的程序编译并提交运行。 调测HDFS应用 查看程序运行结果 程序运行结果会写在用户指定的路径下。用户还可以通过UI查看应用运行情况。 调测HDFS应用 父主题： HDFS开发指南（普通模式）

常用概念 Colocation 同分布（Colocation）功能是将存在关联关系的数据或可能要进行关联操作的数据存储在相同的存储节点上。HDFS文件同分布的特性是，将那些需进行关联操作的文件存放在相同的数据节点上，在进行关联操作计算时，避免了到别的数据节点上获取数据的动作，大大降低了网络带宽的占用。 Client HDFS Client主要包括五种方式：JAVA API、C API、Shell、HTTP REST API、WEB UI五种方式，可参考HDFS常用API介绍、HDFS Shell命令介绍。 JAVA API 提供HDFS文件系统的应用接口，本开发指南主要介绍如何使用Java API进行HDFS文件系统的应用开发。 C API 提供HDFS文件系统的应用接口，使用C语言开发的用户可参考C接口的描述进行应用开发。 Shell 提供shell命令完成HDFS文件系统的基本操作。 HTTP REST API 提供除Shell、Java API和C API以外的其他接口，可通过此接口监控HDFS状态等信息。 WEB UI 提供Web可视化组件管理界面。

代码样例 以下代码片段在com.huawei.bigdata.hbase.examples包的“GlobalSecondaryIndexSample”类中。 本样例用于查询指定id对应的id、age、name信息，并命中idx_id_age索引，由于查询结果被完全覆盖，且无需回原表查询，能够达到最优的查询性能。 /** * Scan data by secondary index. */public void testScanDataByIndex() { LOG.info("Entering testScanDataByIndex."); Scan scan = new Scan(); // Create a filter for indexed column. SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("id"), CompareOperator.EQUAL, Bytes.toBytes("3")); filter.setFilterIfMissing(true); scan.setFilter(filter); // Specify returned columns // If returned columns not included in index table, will query back user table, // it's not the fast way to get data, suggest to cover all needed columns. // If you want to confirm whether using index for scanning, please set hbase client log level to DEBUG. scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("id")); scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age")); scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name")); LOG.info("Scan indexed data."); try (Table table = conn.getTable(tableName); ResultScanner scanner = table.getScanner(scan)) { for (Result result : scanner) { for (Cell cell : result.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("Scan data by index successfully."); } catch (IOException e) { LOG.error("Scan data by index failed ", e); } LOG.info("Exiting testScanDataByIndex.");}

开发思路 根据前述场景说明进行功能分解，以“/user/hdfs-examples/test.txt”文件的读写删除等操作为例，说明HDFS文件的基本操作流程，可分为以下八部分： 创建FileSystem对象：fSystem。 调用fSystem的mkdir接口创建目录。 调用fSystem的create接口创建FSDataOutputStream对象：out，使用out的write方法写入数据。 调用fSystem的append接口创建FSDataOutputStream对象：out，使用out的write方法追加写入数据。 调用fSystem的open接口创建FSDataInputStream对象：in，使用in的read方法读取文件。 调用fSystem中的delete接口删除文件。 调用fSystem中的delete接口删除文件夹。

Hive WebHCat接口介绍 以下示例的IP为WebHCat的业务IP，端口为安装时设置的WebHCat HTTP端口。 除“:version”、“status”、“version”、“version/hive”、“version/hadoop”以外，其他API都需要添加user.name参数。 :version(GET) 描述 查询WebHCat支持的返回类型列表。 URL http://www.myserver.com/templeton/:version 参数 参数 描述 :version WebHCat版本号（当前必须是v1）。 返回结果 参数 描述 responseTypes WebHCat支持的返回类型列表。 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1' status (GET) 描述 获取当前服务器的状态 URL http://www.myserver.com/templeton/v1/status 参数 无 返回结果 参数 描述 status WebHCat连接正常，返回OK。 version 字符串，包含版本号，比如v1。 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/status' version (GET) 描述 获取服务器WebHCat的版本 URL http://www.myserver.com/templeton/v1/version 参数 无 返回结果 参数 描述 supportedVersions 所有支持的版本。 version 当前服务器WebHCat的版本。 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/version' version/hive (GET) 描述 获取服务器Hive的版本 URL http://www.myserver.com/templeton/v1/version/hive 参数 无 返回结果 参数 描述 module hive version Hive的版本 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/version/hive' version/hadoop (GET) 描述 获取服务器Hadoop的版本 URL http://www.myserver.com/templeton/v1/version/hadoop 参数 无 返回结果 参数 描述 module hadoop version Hadoop的版本 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/version/hadoop' ddl (POST) 描述 执行DDL语句 URL http://www.myserver.com/templeton/v1/ddl 参数 参数 描述 exec 需要执行的HCatalog DDL语句。 group 当DDL是创建表时，创建表使用的用户组。 permissions 当DDL是创建表时，创建表使用的权限，格式为rwxr-xr-x。 返回结果 参数 描述 stdout HCatalog执行时的标准输出值，可能为空。 stderr HCatalog执行时的错误输出，可能为空。 exitcode HCatalog的返回值。 例子 curl -i -u : --negotiate -d exec="show tables" 'http://10.64.35.144:9111/templeton/v1/ddl?user.name=user1' ddl/database (GET) 描述 列出所有的数据库 URL http://www.myserver.com/templeton/v1/ddl/database 参数 参数 描述 like 用来匹配数据库名的正则表达式。 返回结果 参数 描述 databases 数据库名 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/ddl/database?user.name=user1' ddl/database/:db (GET) 描述 获取指定数据库的详细信息 URL http://www.myserver.com/templeton/v1/ddl/database/:db 参数 参数 描述 :db 数据库名 返回结果 参数 描述 location 数据库位置 comment 数据库的备注，如果没有备注则不存在 database 数据库名 owner 数据库的所有者 owertype 数据库所有者的类型 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/ddl/database/default?user.name=user1' ddl/database/:db (PUT) 描述 创建数据库 URL http://www.myserver.com/templeton/v1/ddl/database/:db 参数 参数 描述 :db 数据库名 group 创建数据库时使用的用户组 permission 创建数据库时使用的权限 location 数据库的位置 comment 数据库的备注，比如描述 properties 数据库属性 返回结果 参数 描述 database 新创建的数据库的名字 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{"location": "/tmp/a", "comment": "my db", "properties": {"a": "b"}}' 'http://10.64.35.144:9111/templeton/v1/ddl/database/db2?user.name=user1' ddl/database/:db (DELETE) 描述 删除数据库 URL http://www.myserver.com/templeton/v1/ddl/database/:db 参数 参数 描述 :db 数据库名 ifExists 如果指定数据库不存在，Hive会返回错误，除非设置了ifExists为true。 option 将参数设置成cascade或者restrict。如果选择cascade，将清除一切，包括数据和定义。如果选择restrict，表格内容为空，模式也将不存在。 返回结果 参数 描述 database 删除的数据库名字 例子 curl -i -u : --negotiate -X DELETE 'http://10.64.35.144:9111/templeton/v1/ddl/database/db3?ifExists=true&user.name=user1' ddl/database/:db/table (GET) 描述 列出数据库下的所有表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table 参数 参数 描述 :db 数据库名 like 用来匹配表名的正则表达式 返回结果 参数 描述 database 数据库名字 tables 数据库下表名列表 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/ddl/database/default/table?user.name=user1' ddl/database/:db/table/:table (GET) 描述 获取表的详细信息 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table 参数 参数 描述 :db 数据库名 :table 表名 format 格式：format=extended，参考额外信息（“table extended like”）。 返回结果 参数 描述 columns 列名和类型 database 数据库名 table 表名 partitioned 是否分区表，只有extended下才会显示。 location 表的位置，只有extended下才会显示。 outputformat 输出形式，只有extended下才会显示。 inputformat 输入形式，只有extended下才会显示。 owner 表的属主，只有extended下才会显示。 partitionColumns 分区的列，只有extended下才会显示。 例子 curl -i -u : --negotiate 'http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1?format=extended&user.name=user1' ddl/database/:db/table/:table (PUT) 描述 创建表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table 参数 参数 描述 :db 数据库名 :table 新建表名 group 创建表时使用的用户组 permissions 创建表时使用的权限 external 指定位置，hive不使用表的默认位置。 ifNotExists 设置为true，当表存在时不会报错。 comment 备注 columns 列描述，包括列名，类型和可选备注。 partitionedBy 分区列描述，用于划分表格。参数columns列出了列名，类型和可选备注。 clusteredBy 分桶列描述，参数包括columnNames、sortedBy、和numberOfBuckets。参数columnNames包括columnName和排列顺序（ASC为升序，DESC为降序）。 format 存储格式，参数包括rowFormat，storedAs，和storedBy。 location HDFS路径 tableProperties 表属性和属性值（name-value对） 返回结果 参数 描述 database 数据库名 table 表名 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{"columns": [{"name": "id", "type": "int"}, {"name": "name","type": "string"}], "comment": "hello","format": {"storedAs": "orc"} }' 'http://10.64.35.144:9111/templeton/v1/ddl/database/db3/table/tbl1?user.name=user1' ddl/database/:db/table/:table (POST) 描述 重命名表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table 参数 参数 描述 :db 数据库名 :table 已有表名 rename 新表表名 返回结果 参数 描述 database 数据库名 table 新表表名 例子 curl -i -u : --negotiate -d rename=table1 'http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/tbl1?user.name=user1' ddl/database/:db/table/:table (DELETE) 描述 删除表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table 参数 参数 描述 :db 数据库名 :table 表名 ifExists 当设置为true时，不报错。 返回结果 参数 描述 database 数据库名 table 表名 例子 curl -i -u : --negotiate -X DELETE 'http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/table2?ifExists=true&user.name=user1' ddl/database/:db/table/:existingtable/like/:newtable (PUT) 描述 创建一张和已经存在的表一样的表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:existingtable/like/:newtable 参数 参数 描述 :db 数据库名 :existingtable 已有表名 :newtable 新表名 group 创建表时使用的用户组。 permissions 创建表时使用的权限。 external 指定位置，hive不使用表的默认位置。 ifNotExists 当设置为true时，如果表已经存在，Hive不报错。 location HDFS路径 返回结果 参数 描述 database 数据库名 table 表名 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{"ifNotExists": "true"}' 'http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/like/tt1?user.name=user1' ddl/database/:db/table/:table/partition(GET) 描述 列出表的分区信息 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/partition 参数 参数 描述 :db 数据库名 :table 表名 返回结果 参数 描述 database 数据库名 table 表名 partitions 分区属性值和分区名 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/x1/partition?user.name=user1 ddl/database/:db/table/:table/partition/:partition(GET) 描述 列出表的某个具体分区的信息 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/partition/:partition 参数 参数 描述 :db 数据库名 :table 表名 :partition 分区名，解码http引用时，需当心。比如country=%27algeria%27。 返回结果 参数 描述 database 数据库名 table 表名 partition 分区名 partitioned 如果设置为true，为分区表 location 表的存储路径 outputFormat 输出格式 columns 列名，类型，备注 owner 所有者 partitionColumns 分区的列 inputFormat 输入格式 totalNumberFiles 分区下文件个数 totalFileSize 分区下文件总大小 maxFileSize 最大文件大小 minFileSize 最小文件大小 lastAccessTime 最后访问时间 lastUpdateTime 最后更新时间 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/x1/partition/dt=1?user.name=user1 ddl/database/:db/table/:table/partition/:partition(PUT) 描述 增加一个表分区 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/partition/:partition 参数 参数 描述 :db 数据库名。 :table 表名。 group 创建新分区时使用的用户组。 permissions 创建新分区时用户的权限。 location 新分区的存放位置。 ifNotExists 如果设置为true， 当分区已经存在，系统报错。 返回结果 参数 描述 database 数据库名 table 表名 partitions 分区名 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{}' http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/x1/partition/dt=10?user.name=user1 ddl/database/:db/table/:table/partition/:partition(DELETE) 描述 删除一个表分区 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/partition/:partition 参数 参数 描述 :db 数据库名。 :table 表名。 group 删除新分区时使用的用户组。 permissions 删除新分区时用户的权限， 格式为rwxrw-r-x。 ifExists 如果指定分区不存在，Hive报错。参数值设置为true除外。 返回结果 参数 描述 database 数据库名 table 表名 partitions 分区名 例子 curl -i -u : --negotiate -X DELETE -HContent-type:application/json -d '{}' http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/x1/partition/dt=10?user.name=user1 ddl/database/:db/table/:table/column(GET) 描述 获取表的column列表 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/column 参数 参数 描述 :db 数据库名 :table 表名 返回结果 参数 描述 database 数据库名 table 表名 columns 列名字和类型 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/column?user.name=user1 ddl/database/:db/table/:table/column/:column(GET) 描述 获取表的某个具体的column的信息 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/column/:column 参数 参数 描述 :db 数据库名 :table 表名 :column 列名 返回结果 参数 描述 database 数据库名 table 表名 column 列名字和类型 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/column/id?user.name=user1 ddl/database/:db/table/:table/column/:column(PUT) 描述 增加表的一列 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/column/:column 参数 参数 描述 :db 数据库名 :table 表名 :column 列名 type 列类型，比如string和int comment 列备注，比如描述 返回结果 参数 描述 database 数据库名 table 表名 column 列名 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{"type": "string", "comment": "new column"}' http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/column/name?user.name=user1 ddl/database/:db/table/:table/property(GET) 描述 获取表的property URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/property 参数 参数 描述 :db 数据库名 :table 表名 返回结果 参数 描述 database 数据库名 table 表名 properties 属性列表 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/property?user.name=user1 ddl/database/:db/table/:table/property/:property(GET) 描述 获取表的某个具体的property的值 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/property/:property 参数 参数 描述 :db 数据库名 :table 表名 :property 属性名 返回结果 参数 描述 database 数据库名 table 表名 property 属性列表 例子 curl -i -u : --negotiate http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/property/last_modified_by?user.name=user1 ddl/database/:db/table/:table/property/:property(PUT) 描述 增加表的property的值 URL http://www.myserver.com/templeton/v1/ddl/database/:db/table/:table/property/:property 参数 参数 描述 :db 数据库名 :table 表名 :property 属性名 value 属性值 返回结果 参数 描述 database 数据库名 table 表名 property 属性名 例子 curl -i -u : --negotiate -X PUT -HContent-type:application/json -d '{"value": "my value"}' http://10.64.35.144:9111/templeton/v1/ddl/database/default/table/t1/property/mykey?user.name=user1 mapreduce/jar(POST) 描述 执行MR任务，在执行之前，需要将MR的jar包上传到HDFS中 URL http://www.myserver.com/templeton/v1/mapreduce/jar 参数 参数 描述 jar 需要执行的MR的jar包。 class 需要执行的MR的分类。 libjars 需要加入的classpath的jar包名，以逗号分隔。 files 需要复制到集群的文件名，以逗号分隔。 arg Main类接受的输入参数。 define 设置hadoop的配置，格式为：define=NAME=VALUE。 statusdir WebHCat会将执行的MR任务的状态写入到statusdir中。如果设置了这个值，那么需要用户手动进行删除。 enablelog 如果statusdir设置，enablelog设置为true，收集Hadoop任务配置和日志到$statusdir/logs。此后，成功和失败的尝试，都将记录进日志。$statusdir/logs下，子目录布局为： logs/$job_id (directory for $job_id) logs/$job_id/job.xml.html logs/$job_id/$attempt_id (directory for $attempt_id) logs/$job_id/$attempt_id/stderr logs/$job_id/$attempt_id/stdout logs/$job_id/$attempt_id/syslog 仅支持Hadoop 1.X。 callback 在MR任务执行完的回调地址，使用$jobId，将任务ID嵌入回调地址。在回调地址中，任务ID替换该$jobId。 返回结果 参数 描述 id 任务ID，类似“job_201110132141_0001” 例子 curl -i -u : --negotiate -d jar="/tmp/word.count-0.0.1-SNAPSHOT.jar" -d class=com.huawei.word.count.WD -d statusdir="/output" "http://10.64.35.144:9111/templeton/v1/mapreduce/jar?user.name=user1" mapreduce/streaming(POST) 描述 以Streaming方式提交MR任务 URL http://www.myserver.com/templeton/v1/mapreduce/streaming 参数 参数 描述 input Hadoop中input的路径。 output 存储output的路径。如没有规定，WebHCat将output储存在使用队列资源可以发现到的路径。 mapper mapper程序位置。 reducer reducer程序位置。 files HDFS文件添加到分布式缓存中。 arg 设置argument。 define 设置hadoop的配置变量，格式：define=NAME=VALUE cmdenv 设置环境变量，格式：cmdenv=NAME=VALUE statusdir WebHCat会将执行的MR任务的状态写入到statusdir中。如果设置了这个值，那么需要用户手动进行删除。 enablelog 如果statusdir设置，enablelog设置为true，收集Hadoop任务配置和日志到$statusdir/logs。此后，成功和失败的尝试，都将记录进日志。$statusdir/logs下，子目录布局为： logs/$job_id (directory for $job_id) logs/$job_id/job.xml.html logs/$job_id/$attempt_id (directory for $attempt_id) logs/$job_id/$attempt_id/stderr logs/$job_id/$attempt_id/stdout logs/$job_id/$attempt_id/syslog 仅支持Hadoop 1.X。 callback 在MR任务执行完的回调地址，使用$jobId，将任务ID嵌入回调地址。在回调地址中，任务ID将替换该$jobId。 返回结果 参数 描述 id 任务ID，类似job_201110132141_0001 例子 curl -i -u : --negotiate -d input=/input -d output=/oooo -d mapper=/bin/cat -d reducer="/usr/bin/wc -w" -d statusdir="/output" 'http://10.64.35.144:9111/templeton/v1/mapreduce/streaming?user.name=user1' 本接口的使用需要前置条件，请参阅“开发规范”中的Hive规则。 /hive(POST) 描述 执行Hive命令 URL http://www.myserver.com/templeton/v1/hive 参数 参数 描述 execute hive命令，包含整个和短的Hive命令。 file 包含hive命令的HDFS文件。 files 需要复制到集群的文件名，以逗号分隔。 arg 设置argument。 define 设置Hive的配置，格式：define=key=value，如果使用多实例，需要配置实例的scratch dir，如WebHCat实例使用define=hive.exec.scratchdir=/tmp/hive-scratch，WebHCat1实例使用define=hive.exec.scratchdir=/tmp/hive1-scratch，以此类推。 statusdir WebHCat会将执行的MR任务的状态写入到statusdir中。如果设置了这个值，那么需要用户手动进行删除。 enablelog 如果statusdir设置，enablelog设置为true，收集Hadoop任务配置和日志到$statusdir/logs。此后，成功和失败的尝试，都将记录进日志。$statusdir/logs下，子目录布局为： logs/$job_id (directory for $job_id) logs/$job_id/job.xml.html logs/$job_id/$attempt_id (directory for $attempt_id) logs/$job_id/$attempt_id/stderr logs/$job_id/$attempt_id/stdout logs/$job_id/$attempt_id/syslog callback 在MR任务执行完的回调地址，使用$jobId，将任务ID嵌入回调地址。在回调地址中，任务ID将替换该$jobId。 返回结果 参数 描述 id 任务ID，类似job_201110132141_0001 例子 curl -i -u : --negotiate -d execute="select count(*) from t1" -d statusdir="/output" -d define=hive.exec.scratchdir=/tmp/hive-scratch "http://10.64.35.144:9111/templeton/v1/hive?user.name=user1" jobs(GET) 描述 获取所有的job id URL http://www.myserver.com/templeton/v1/jobs 参数 参数 描述 fields 如果设置成*，那么会返回每个job的详细信息。如果没设置，只返回任务ID。现在只能设置成*，如设置成其他值，将出现异常。 jobid 如果设置了jobid，那么只有字典顺序比jobid大的job才会返回。比如，如果jobid为"job_201312091733_0001"，只有大于该值的job才能返回。返回的job的个数，取决于numrecords。 numrecords 如果设置了numrecords和jobid，jobid列表按字典顺序排列，待jobid返回后，可以得到numrecords的最大值。如果jobid没有设置， 而numrecords设置了参数值，jobid按字典顺序排列后，可以得到numrecords的最大值。相反，如果numrecords没有设置，而jobid设置了参数值，所有大于jobid的job都将返回。 showall 如果设置为true，用户可以获取所有job，如果设置为false，则只获取当前用户提交的job。默认为false。 返回结果 参数 描述 id Job id detail 如果showall为true，那么显示detail信息，否则为null。 例子 curl -i -u : --negotiate "http://10.64.35.144:9111/templeton/v1/jobs?user.name=user1" jobs/:jobid(GET) 描述 获取指定job的信息 URL http://www.myserver.com/templeton/v1/jobs/:jobid 参数 参数 描述 jobid Job创建后的Jobid 返回结果 参数 描述 status 包含job状态信息的json对象。 profile 包含job状态的json对象。WebHCat解析JobProfile对象中的信息，该对象因Hadoop版本不同而不同。 id Job的id。 percentComplete 完成百分比，比如75% complete，如果完成后则为null。 user 创建job的用户。 callback 回调URL（如果有）。 userargs 用户提交job时的argument参数和参数值。 exitValue job退出值。 例子 curl -i -u : --negotiate "http://10.64.35.144:9111/templeton/v1/jobs/job_1440386556001_0255?user.name=user1" jobs/:jobid(DELETE) 描述 kill任务 URL http://www.myserver.com/templeton/v1/jobs/:jobid 参数 参数 描述 :jobid 删除的Job的ID 返回结果 参数 描述 user 提交Job的用户。 status 包含Job状态信息的JSON对象。 profile 包含Job信息的JSON对象。WebHCat解析JobProfile对象中的信息，该对象因Hadoop版本不同而不同。 id Job的ID。 callback 回调的URL（如果有）。 例子 curl -i -u : --negotiate -X DELETE "http://10.64.35.143:9111/templeton/v1/jobs/job_1440386556001_0265?user.name=user1" 父主题： Hive对外接口介绍

操作步骤 参考导入并配置HetuEngine Python3样例工程章节，获取样例代码，获取hetu-jdbc-XXX.jar文件，并复制到自定义目录中。 编辑样例代码，根据集群实际情况修改url、user等信息，并根据实际路径修改“jdbc_location”。 Windows系统路径填写示例："D:\\hetu-examples-python3\\hetu-jdbc-XXX.jar" Linux系统路径示例："/opt/hetu-examples-python3/hetu-jdbc-XXX.jar" 运行python3样例代码。 Windows直接通过pycharm或者Python IDLE运行py脚本。 Linux运行样例代码前需要已安装Java。 进入样例代码路径并执行py脚本，样例代码路径如“/opt/hetu-examples-python3”： cd /opt/hetu-examples-python3 python3 JDBCExampleBroker.py 运行结果。 Windows系统中，在console界面可以看到运行结果示例： 图1 Windows系统运行结果 Linux系统中运行结果示例： …[ ('hive', ('mv',), ('svc',), ('system',), ('systemremote',)]

场景说明 假定用户有某个网站周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Flink的DataStream应用程序实现如下功能： DataStream应用程序可以在Windows环境和Linux环境中运行。 实时统计总计网购时间超过2个小时的女性网民信息。 周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。 log1.txt：周六网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20YuanJing,male,10GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20FangBo,female,50LiuYang,female,20YuanJing,male,10GuoYijun,male,50CaiXuyu,female,50FangBo,female,60 log2.txt：周日网民停留日志。该日志文件在该样例程序中的data目录下获取。 LiuYang,female,20YuanJing,male,10CaiXuyu,female,50FangBo,female,50GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20CaiXuyu,female,50FangBo,female,50LiuYang,female,20YuanJing,male,10FangBo,female,50GuoYijun,male,50CaiXuyu,female,50FangBo,female,60

数据规划 DataStream样例工程的数据存储在文本中。 将log1.txt和log2.txt放置在指定路径下，例如"/opt/log1.txt"和"/opt/log2.txt"。 数据文件若存放在本地文件系统，需在所有部署Yarn NodeManager的节点指定目录放置，并设置运行用户访问权限。 或将数据文件放置于HDFS，并指定程序中读取文件路径HDFS路径，例如"hdfs://hacluster/path/to/file"。

问题 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.

代码样例 下面代码片段在com.huawei.bigdata.hbase.examples包的“TestMain”类的init方法中。 private static void init() throws IOException { // Default load from conf directory conf = HBaseConfiguration.create(); //In Windows environment 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”的目录下。

代码样例 以下代码片段在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."); }

准备开发环境 在进行二次开发时，要准备的开发和运行环境如表1所示。 表1 开发环境 准备项 说明 操作系统 开发环境：Windows系统，支持Windows 7以上版本。 运行环境：Windows系统或Linux系统。 如需在本地调测程序，运行环境需要和集群业务平面网络互通。 安装JDK 开发和运行环境的基本配置，版本要求如下： 服务端和客户端仅支持自带的OpenJDK，版本为1.8.0_272，不允许替换。 对于客户应用需引用SDK类的Jar包运行在客户应用进程中的。 X86客户端：Oracle JDK：支持1.8版本；IBM JDK：支持1.8.5.11版本。 TaiShan客户端：OpenJDK：支持1.8.0_272版本。 说明： 基于安全考虑，服务端只支持TLS V1.2及以上的加密协议。 IBM JDK默认只支持TLS V1.0，若使用IBM JDK，请配置启动参数“com.ibm.jsse2.overrideDefaultTLS”为“true”，设置后可以同时支持TLS V1.0/V1.1/V1.2，详情参见https://www.ibm.com/support/knowledgecenter/zh/SSYKE2_8.0.0/com.ibm.java.security.component.80.doc/security-component/jsse2Docs/matchsslcontext_tls.html#matchsslcontext_tls。 安装和配置IntelliJ IDEA 用于开发HBase应用程序的工具。版本要求：2019.1或其他兼容版本。 说明： 若使用IBM JDK，请确保IntelliJ IDEA中的JDK配置为IBM JDK。 若使用Oracle JDK，请确保IntelliJ IDEA中的JDK配置为Oracle JDK。 若使用Open JDK，请确保IntelliJ IDEA中的JDK配置为Open JDK。 安装Junit插件 开发环境的基本配置。 安装Maven 开发环境的基本配置。用于项目管理，贯穿软件开发生命周期。 7-zip 用于解压“*.zip”和“*.rar”文件。 支持7-Zip 16.04版本。

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

操作步骤 以客户端安装用户，登录安装HBase客户端的节点。 进入HBase客户端安装目录： 例如：cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户，当前用户需要具有创建HBase表的权限，具体请参见创建角色配置拥有对应权限的角色，参考创建用户为用户绑定对应角色。如果当前集群未启用Kerberos认证，则无需执行此命令。 kinit MRS 集群用户 例如，kinit hbaseuser。 直接执行Phoenix客户端命令。 sqlline.py 建表： CREATE TABLE TEST (id VARCHAR PRIMARY KEY, name VARCHAR); 插入数据： UPSERT INTO TEST(id,name) VALUES ('1','jamee'); 查询数据： SELECT * FROM TEST; 删表： DROP TABLE TEST; 退出Phoenix命令行。 !quit

回答 建议将"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正常启动并且成功创建目录。

开发思路 数据准备。 创建三张表，雇员信息表“employees_info”、雇员联络信息表“employees_contact”、雇员信息扩展表“employees_info_extended”。 雇员信息表“employees_info”的字段为雇员编号、姓名、支付薪水币种、薪水金额、缴税税种、工作地、入职时间，其中支付薪水币种“R”代表人民币，“D”代表美元。 雇员联络信息表“employees_contact”的字段为雇员编号、电话号码、e-mail。 雇员信息扩展表“employees_info_extended”的字段为雇员编号、姓名、电话号码、e-mail、支付薪水币种、薪水金额、缴税税种、工作地，分区字段为入职时间。 创建表代码实现请见创建Hive表。 加载雇员信息数据到雇员信息表“employees_info”中。 加载数据代码实现请见加载数据到Hive表中。 雇员信息数据如表1所示： 表1 雇员信息数据 编号 姓名 支付薪水币种 薪水金额 缴税税种 工作地 入职时间 1 Wang R 8000.01 personal income tax&0.05 Country1:City1 2014 3 Tom D 12000.02 personal income tax&0.09 Country2:City2 2014 4 Jack D 24000.03 personal income tax&0.09 Country3:City3 2014 6 Linda D 36000.04 personal income tax&0.09 Country4:City4 2014 8 Zhang R 9000.05 personal income tax&0.05 Country5:City5 2014 加载雇员联络信息数据到雇员联络信息表“employees_contact”中。 雇员联络信息数据如表2所示： 表2 雇员联络信息数据 编号 电话号码 e-mail 1 135 XXXX XXXX xxxx@xx.com 3 159 XXXX XXXX xxxxx@xx.com.cn 4 186 XXXX XXXX xxxx@xx.org 6 189 XXXX XXXX xxxx@xxx.cn 8 134 XXXX XXXX xxxx@xxxx.cn 数据分析。 数据分析代码实现，请见查询Hive表数据。 查看薪水支付币种为美元的雇员联系方式。 查询入职时间为2014年的雇员编号、姓名等字段，并将查询结果加载进表employees_info_extended中的入职时间为2014的分区中。 统计表employees_info中有多少条记录。 查询使用以“cn”结尾的邮箱的员工信息。 提交数据分析任务，统计表employees_info中有多少条记录。