华为云用户手册

在本地Windows环境中查看Spark程序调试结果 SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/D:/mavenlocal/org/apache/logging/log4j/log4j-slf4j-impl/2.6.2/log4j-slf4j-impl-2.6.2.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/D:/mavenlocal/org/slf4j/slf4j-log4j12/1.7.30/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation. SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory] ERROR StatusLogger No log4j2 configuration file found. Using default configuration: logging only errors to the console. ---- Begin executing sql: CREATE TABLE IF NOT EXISTS CHILD (NAME STRING, AGE INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' ---- Result ---- Done executing sql: CREATE TABLE IF NOT EXISTS CHILD (NAME STRING, AGE INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' ---- ---- Begin executing sql: LOAD DATA INPATH 'hdfs:/data/data' INTO TABLE CHILD ---- Result ---- Done executing sql: LOAD DATA INPATH 'hdfs:/data/data' INTO TABLE CHILD ---- ---- Begin executing sql: SELECT * FROM child ---- NAME AGE Miranda 32 Karlie 23 Candice 27 ---- Done executing sql: SELECT * FROM child ---- ---- Begin executing sql: DROP TABLE child ---- Result ---- Done executing sql: DROP TABLE child ---- Process finished with exit code 0 父主题： 在本地Windows环境中调测Spark应用

回答 Spark任务在运行过程中，driver会创建一个spark-开头的本地临时目录，用于存放业务jar包，配置文件等，同时在本地创建一个blockmgr-开头的本地临时目录，用于存放block data。此两个目录会在Spark应用运行结束时自动删除。 此两个目录的存放路径优先通过SPARK_LOCAL_DIRS环境变量指定，若不存在该环境变量，则设置为spark.local.dir的值，若此配置还不存在，则使用java.io.tmpdir的值。客户端默认配置中spark.local.dir被设置为/tmp，因此默认使用系统/tmp目录。 但存在一些特殊情况，如driver进程未正常退出，比如被kill -9命令结束进程，或者Java虚拟机直接崩溃等场景，导致driver的退出流程未正常执行，则可能导致该部分目录无法被正常清理，残留在系统中。 当前只有yarn-client模式和local模式的driver进程会产生上述问题，在yarn-cluster模式中，已将container内进程的临时目录设置为container临时目录，当container退出时，由container自动清理该目录，因此yarn-cluster模式不存在此问题。

操作步骤 修改WordCountTopology.java类，使用remoteSubmit方式提交应用程序。并替换用户keytab文件名称，用户principal名称，和Jar文件地址。 使用remoteSubmit方式提交应用程序 public static void main(String[] args) throws Exception { TopologyBuilder builder = buildTopology(); /* * 任务的提交认为三种方式 * 1、命令行方式提交，这种需要将应用程序jar包复制到客户端机器上执行客户端命令提交 * 2、远程方式提交，这种需要将应用程序的jar包打包好之后在IntelliJ IDEA中运行main方法提交 * 3、本地提交 ，在本地执行应用程序，一般用来测试 * 命令行方式和远程方式安全和普通模式都支持 * 本地提交仅支持普通模式 * * 用户同时只能选择一种任务提交方式，默认命令行方式提交，如果是其他方式，请删除代码注释即可 */ submitTopology(builder, SubmitType.REMOTE); } 根据实际情况修改userJarFilePath为实际的拓扑Jar包地址 private static void remoteSubmit(TopologyBuilder builder) throws AlreadyAliveException, InvalidTopologyException, NotALeaderException, AuthorizationException, IOException { Config config = createConf(); String userJarFilePath = "D:\\example.jar"; System.setProperty(STORM_SUBMIT_JAR_PROPERTY, userJarFilePath); //安全模式下的一些准备工作 if (isSecurityModel()) { securityPrepare(config); } config.setNumWorkers(1); StormSubmitter.submitTopologyWithProgressBar(TOPO LOG Y_NAME, config, builder.createTopology()); } 安全模式下需要执行安全准备，根据实际情况修改userKeyTablePath和userPrincipal为导入并配置Storm样例工程章节的步骤2中所获取用户的keytab文件路径和principal private static void securityPrepare(Config config) throws IOException { String userKeyTablePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator + "user.keytab"; String userPrincipal = "StreamingDeveloper"; String krbFilePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator +"krb5.conf"; //windows路径下分隔符替换 userKeyTablePath = userKeyTablePath.replace("\\", "\\\\"); krbFilePath = krbFilePath.replace("\\", "\\\\"); String principalInstance = String.valueOf(config.get(Config.STORM_SECURITY_PRINCIPAL_INSTANCE)); LoginUtil.setKrb5Config(krbFilePath); LoginUtil.setZookeeperServerPrincipal("zookeeper/" + principalInstance); LoginUtil.setJaasFile(userPrincipal, userKeyTablePath); } 执行WordCountTopology.java类的Main方法提交应用程序。

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

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中编包并运行Spark程序。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中的HBaseSource文件： # -*- coding:utf-8 -*- """ 【说明】 由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现 """ from py4j.java_gateway import java_import from pyspark.sql import SparkSession # 创建SparkSession spark = SparkSession\ .builder\ .appName("HBaseSourceExample")\ .getOrCreate() # 向sc._jvm中导入要运行的类 java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.datasources.HBaseSource') # 创建类实例并调用方法，传递sc._jsc参数 spark._jvm.HBaseSource().execute(spark._jsc) # 停止SparkSession spark.stop()

提交命令 假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端“$SPARK_HOME”目录下，以下命令均在“$SPARK_HOME”目录执行。 yarn-client模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --jars /opt/female/protobuf-java-2.5.0.jar --conf spark.yarn.user.classpath.first=true --class com.huawei.bigdata.spark.examples.datasources.HBaseSource SparkOnHbaseJavaExample-1.0.jar python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --conf spark.yarn.user.classpath.first=true --jars SparkOnHbaseJavaExample-1.0.jar,/opt/female/protobuf-java-2.5.0.jar HBaseSource.py yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --jars /opt/female/protobuf-java-2.5.0.jar --conf spark.yarn.user.classpath.first=true --class com.huawei.bigdata.spark.examples.datasources.HBaseSource SparkOnHbaseJavaExample-1.0.jar python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --conf spark.yarn.user.classpath.first=true --jars SparkOnHbaseJavaExample-1.0.jar,/opt/female/protobuf-java-2.5.0.jar HBaseSource.py

解决步骤 检查工程conf目录下“producer.properties”中配置的“bootstrap.servers”配置值中访问的IP和端口是否正确： 如果IP与Kafka集群部署的业务IP不一致，那么需要修改为当前集群正确的IP地址。 如果配置中的端口为21007（Kafka安全模式端口），那么修改该端口为9092（Kafka普通模式端口）。 检查网络是否正常，确保当前机器能够正常访问Kafka集群。

代码样例 下面代码片段在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; //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); }

Hive应用开发流程 开发流程中各阶段的说明如图1和表1所示。 图1 Hive应用程序开发流程 表1 Hive应用开发的流程说明 阶段 说明 参考文档 准备开发环境 在进行应用开发前，需首先准备开发环境，推荐使用Java语言进行开发，使用IntelliJ IDEA工具，同时完成JDK、Maven等初始配置。 准备本地应用开发环境 准备连接集群配置文件 应用程序开发或运行过程中，需通过集群相关配置文件信息连接 MRS 集群，配置文件通常包括集群组件信息文件，可从已创建好的MRS集群中获取相关内容。 用于程序调测或运行的节点，需要与MRS集群内节点网络互通，同时配置hosts 域名 信息。 准备连接Hive集群配置文件 配置并导入样例工程 Hive提供了不同场景下的多种样例程序，用户可获取样例工程并导入本地开发环境中进行程序学习。 导入并配置Hive样例工程 根据业务场景开发程序 根据实际业务场景开发程序，调用组件接口实现对应功能。 开发Hive应用 编译并运行程序 开发好的程序编译运行，用户可在本地Windows开发环境中进行程序调测运行，也可以将程序编译为Jar包后，提交到Linux节点上运行。 调测Hive应用 父主题： Hive应用开发概述

简介 Storm是一个分布式的、可靠的、容错的数据流处理系统。它会把工作任务委托给不同类型的组件，每个组件负责处理一项简单特定的任务。Storm的目标是提供对大数据流的实时处理，可以可靠地处理无限的数据流。 Storm有很多适用的场景：实时分析、在线机器学习、持续计算和分布式ETL等，易扩展、支持容错，可确保数据得到处理，易于构建和操控。 Storm有如下几个特点： 适用场景广泛 易扩展，可伸缩性高 保证无数据丢失 容错性好 多语言 易于构建和操控

Oozie应用开发常用概念 流程定义文件 描述业务逻辑的XML文件，包括“workflow.xml”、“coordinator.xml”、“bundle.xml”三类，最终由Oozie引擎解析并执行。 流程属性文件 流程运行期间的参数配置文件，对应文件名为“job.properties”，每个流程定义有且仅有一个该属性文件。 keytab文件 存放用户信息的密钥文件。在安全模式下，应用程序采用此密钥文件进行API方式认证。 Client 客户端直接面向用户，可通过Java API、Shell API、 REST API或者Web UI访问Oozie服务端。 Oozie WebUI界面 通过“https://Oozie服务器IP地址:21003/oozie”登录Oozie WebUI界面。 父主题： Oozie应用开发概述

查看调测结果 ClickHouse应用程序运行完成后，可通过以下方式查看程序运行情况： 通过运行结果查看程序运行情况。 通过ClickHouse日志获取应用运行情况。 即查看当前jar文件所在目录的“logs/clickhouse-example.log”日志文件，例如“客户端安装目录/JDBC/logs/clickhouse-example.log”。 jar包运行结果如下： 2021-06-10 20:53:56,028 | INFO | main | Current load balancer is 10.112.17.150:21426 | com.huawei.clickhouse.examples.Util.insertData(Util.java:128) 2021-06-10 20:53:58,247 | INFO | main | Inert batch time is 1442 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:53:59,649 | INFO | main | Inert batch time is 1313 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:05,872 | INFO | main | Inert batch time is 6132 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:10,223 | INFO | main | Inert batch time is 4272 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:11,614 | INFO | main | Inert batch time is 1300 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:12,871 | INFO | main | Inert batch time is 1200 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:14,589 | INFO | main | Inert batch time is 1663 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:16,141 | INFO | main | Inert batch time is 1500 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:17,690 | INFO | main | Inert batch time is 1498 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:19,206 | INFO | main | Inert batch time is 1468 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:145) 2021-06-10 20:54:19,207 | INFO | main | Inert all batch time is 22626 ms | com.huawei.clickhouse.examples.Util.insertData(Util.java:148) 2021-06-10 20:54:19,208 | INFO | main | Current load balancer is 10.112.17.150:21426 | com.huawei.clickhouse.examples.Util.exeSql(Util.java:58) 2021-06-10 20:54:20,231 | INFO | main | Execute query:select * from mutong1.testtb_all order by age limit 10 | com.huawei.clickhouse.examples.Util.exeSql(Util.java:63) 2021-06-10 20:54:21,266 | INFO | main | Execute time is 1035 ms | com.huawei.clickhouse.examples.Util.exeSql(Util.java:67) 2021-06-10 20:54:21,267 | INFO | main | Current load balancer is 10.112.17.150:21426 | com.huawei.clickhouse.examples.Util.exeSql(Util.java:58) 2021-06-10 20:54:21,815 | INFO | main | Execute query:select toYYYYMM(date),count(1) from mutong1.testtb_all group by toYYYYMM(date) order by count(1) DESC limit 10 | com.huawei.clickhouse.examples.Util.exeSql(Util.java:63) 2021-06-10 20:54:22,897 | INFO | main | Execute time is 1082 ms | com.huawei.clickhouse.examples.Util.exeSql(Util.java:67) 2021-06-10 20:54:22,898 | INFO | main | name age date | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,898 | INFO | main | huawei_266 0 2021-12-19 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_2500 0 2021-12-29 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_8980 0 2021-12-16 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_671 0 2021-12-29 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_2225 0 2021-12-12 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_6040 0 2021-12-14 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_7294 0 2021-12-10 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,899 | INFO | main | huawei_1133 0 2021-12-25 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | huawei_3161 0 2021-12-21 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | huawei_3992 0 2021-11-25 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | toYYYYMM(date) count() | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | 201910 2247 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | 202105 2213 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | 201801 2208 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,900 | INFO | main | 201803 2204 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 201810 2167 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 201805 2166 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 201901 2164 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 201908 2145 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 201912 2143 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144) 2021-06-10 20:54:22,901 | INFO | main | 202107 2137 | com.huawei.clickhouse.examples.Demo.queryData(Demo.java:144)

数据规划 首先需要把原日志文件放置在HDFS系统里。 本地新建两个文本文件input_data1.txt和input_data2.txt，将log1.txt中的内容复制保存到input_data1.txt，将log2.txt中的内容复制保存到input_data2.txt。 在HDFS客户端路径下建立一个文件夹，“/tmp/input”，并上传input_data1.txt，input_data2.txt到此目录，命令如下： 在Linux系统HDFS客户端使用命令hadoop fs -mkdir /tmp/input（hdfs dfs命令有同样的作用），创建对应目录。 进入到HDFS客户端下的“/tmp/input”目录，在Linux系统HDFS客户端使用命令hadoop fs -putinput_data1.txt /tmp/input和hadoop fs -putinput_data2.txt /tmp/input，上传数据文件。

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

问题 执行Spark Core应用，尝试收集大量数据到Driver端，当Driver端内存不足时，应用挂起不退出，日志内容如下。 16/04/19 15:56:22 ERROR Utils: Uncaught exception in thread task-result-getter-2 java.lang.OutOfMemoryError: Java heap space at java.lang.reflect.Array.newArray(Native Method) at java.lang.reflect.Array.newInstance(Array.java:75) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71) at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91) at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716) at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Exception in thread "task-result-getter-2" java.lang.OutOfMemoryError: Java heap space at java.lang.reflect.Array.newArray(Native Method) at java.lang.reflect.Array.newInstance(Array.java:75) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71) at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91) at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716) at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)

回答 用户尝试收集大量数据到Driver端，如果Driver端的内存不足以存放这些数据，那么就会抛出OOM(OutOfMemory)的异常，然后Driver端一直在进行GC，尝试回收垃圾来存放返回的数据，导致应用长时间挂起。 解决措施： 如果用户需要在OOM场景下强制将应用退出，那么可以在启动Spark Core应用时，在客户端配置文件“$SPARK_HOME/conf/spark-defaults.conf”中的配置项“spark.driver.extraJavaOptions”中添加如下内容： -XX:OnOutOfMemoryError='kill -9 %p'

Hive应用开发常用概念 keytab文件 存放用户信息的密钥文件。应用程序采用此密钥文件在MRS产品中进行API方式认证。 客户端 客户端直接面向用户，可通过Java API、Thrift API访问服务端进行Hive的相关操作。 HQL语言 Hive Query Language，类SQL语句。 HCatalog HCatalog是建立在Hive元数据之上的一个表信息管理层，吸收了Hive的DDL命令。为Mapreduce提供读写接口，提供Hive命令行接口来进行数据定义和元数据查询。基于MRS的HCatalog功能，Hive、Mapreduce开发人员能够共享元数据信息，避免中间转换和调整，能够提升数据处理的效率。 WebHCat WebHCat运行用户通过Rest API来执行Hive DDL，提交Mapreduce任务，查询Mapreduce任务执行结果等操作。 父主题： Hive应用开发概述

Kudu简介 Kudu是专为Apache Hadoop平台开发的列式存储管理器，具有Hadoop生态系统应用程序的共同技术特性：在通用的商用硬件上运行，可水平扩展，提供高可用性。 Kudu的设计具有以下优点： 能够快速处理OLAP工作负载。 支持与MapReduce，Spark和其他Hadoop生态系统组件集成。 与Apache Impala的紧密集成，使其成为将HDFS与Apache Parquet结合使用的更好选择。 提供强大而灵活的一致性模型，允许您根据每个请求选择一致性要求，包括用于严格可序列化的一致性的选项。 提供同时运行顺序读写和随机读写的良好性能。 易于管理。 高可用性。Master和TServer采用raft算法，该算法可确保只要副本总数的一半以上可用，tablet就可以进行读写操作。例如，如果3个副本中有2个副本或5个副本中有3个副本可用，则tablet可用。即使主tablet出现故障，也可以通过只读的副tablet提供读取服务。 支持结构化数据模型。 通过结合所有以上属性，Kudu的目标是支持在当前Hadoop存储技术上难以实现或无法实现的应用。 Kudu的应用场景有： 需要最终用户立即使用新到达数据的报告型应用。 同时支持大量历史数据查询和细粒度查询的时序应用。 使用预测模型并基于所有历史数据定期刷新预测模型来做出实时决策的应用。

Oozie应用开发流程 本文档主要基于java API对Oozie进行应用开发。 开发流程中各阶段的说明如图1和表1所示。 图1 Oozie应用程序开发流程 表1 Oozie应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解Oozie的基本概念，了解场景需求等。 Oozie应用开发常用概念 准备开发和运行环境 Oozie的应用程序当前推荐使用Java语言进行开发。可使用IDEA工具。 准备本地应用开发环境 准备工程 Oozie提供了不同场景下的样例程序，您可以导入样例工程进行程序学习。 导入并配置Oozie样例工程 根据场景开发工程 提供了Java语言的样例工程。 开发Oozie应用 编译并运行程序 指导用户将开发好的程序编译并提交运行。 调测Oozie应用 查看程序运行结果 程序运行结果会写在用户指定的路径下。用户还可以通过UI查看应用运行情况。 调测Oozie应用 父主题： Oozie应用开发概述

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

注意事项 当前二级索引不支持使用SubstringComparator类定义的对象作为Filter的比较器。 例如，如下示例中的用法当前不支持： Scan scan = new Scan(); filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL); filterList.addFilter(new SingleColumnValueFilter(Bytes .toBytes(columnFamily), Bytes.toBytes(qualifier), CompareOperator.EQUAL, new SubstringComparator(substring))); scan.setFilter(filterList);

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

样例代码 -- 从本地文件系统/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上指定表的目录下。

样例数据 表employees_info的数据如下： 1,Wang,R,8000.01,person&personal^Btype&income^Btax&0.05,Country1:City1,2014 3,Tom,D,12000.02,person&personal^Btype&income^Btax&0.09,Country2:City2,2014 4,Jack,D,24000.03,person&personal^Btype&income^Btax&0.05,Country3:City3,2014 6,Linda,D,36000.04,person&personal^Btype&income^Btax&0.05,Country4:City4,2014 8,Zhang,R,9000.05,person&personal^Btype&income^Btax&0.05,Country5:City5,2014 表employees_contact的数据如下： 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 表employees_info_extended的数据如下： 1,Wang,135 XXXX XXXX,xxxx@xx.com,R,8000.01,person&personal^Btype&income^Btax&0.05,Country1:City1,2014 3,Tom,159 XXXX XXXX,xxxxx@xx.com.cn,D,12000.02,person&personal^Btype&income^Btax&0.09,Country2:City2,2014 4,Jack,186 XXXX XXXX,xxxx@xx.org,D,24000.03,person&personal^Btype&income^Btax&0.05,Country3:City3,2014 6,Linda,189 XXXX XXXX,xxxx@xxx.cn,D,36000.04,person&personal^Btype&income^Btax&0.05,Country4:City4,2014 8,Zhang,134 XXXX XXXX,xxxx@xxxx.cn,R,9000.05,person&personal^Btype&income^Btax&0.05,Country5:City5,2014

功能介绍 本小节介绍了如何使用HQL向已有的表employees_info中加载数据。从本节中可以掌握如何从本地文件系统、MRS集群中加载数据。以关键字LOCAL区分数据源是否来自本地。 在启用了安全服务的集群中执行如下操作，需要在数据库中具有UPDATE权限及对加载数据文件具有owner权限和读写权限，详情请参见Hive应用开发概述。 如果加载数据语句中有关键字LOCAL，表明从本地加载数据，除要求对相应表的UPDATE权限外，还要求该数据在当前连接的HiveServer节点上，加载用户对数据路径“PATH”具有读权限，且以omm用户能够访问该数据文件。 如果加载数据语句中有关键字OVERWRITE，表示加载的数据会覆盖表中原有的数据，否则加载的数据会追加到表中。

回答 首先查看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。 重新提交任务。

回答 问题原因： 在IBM JDK下建立的JDBC connection时间超过登录用户的认证超时时间（默认一天），导致认证失败。 IBM JDK的机制跟Oracle JDK的机制不同，IBM JDK在认证登录后的使用过程中做了时间检查却没有检测外部的时间更新，导致即使显式调用relogin也无法得到刷新。 解决措施： 通常情况下，在发现JDBC connection不可用的时候，可以关闭该connection，重新创建一个connection继续执行。

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对外接口介绍

部署运行及结果查看 导出本地jar包，请参见打包Storm样例工程应用。 将4中获取的配置文件和5中获取的jar包合并统一打出完整的业务jar包，请参见打包Storm业务。 将开发好的yaml文件及相关的properties文件复制至storm客户端所在主机的任意目录下，如“/opt”。 执行命令提交拓扑。 storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --remote /opt/my-topology.yaml 如果设置业务以本地模式启动，则提交命令如下： storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --local /opt/my-topology.yaml 如果业务设置为本地模式，请确保提交环境为普通模式环境，当前不支持安全环境下使用命令提交本地模式的业务。 如果使用了properties文件，则提交命令如下： storm jar /opt/jartarget/source.jar org.apache.storm.flux.Flux --remote /opt/my-topology.yaml --filter /opt/my-prop.properties 拓扑提交成功后请自行登录storm UI查看。