前言Kylin 用户在使用 Spark的过程中经常会遇到任务提交缓慢、构建节点不稳定的问题。为了更方便地向 Spark 提交、管理和监控任务有些用户会使用 Livy 作为 Spark 的交互接口。在最新的 Apache Kylin 3.0 版本中Kylin 加入了通过 Apache Livy 递交 Spark 任务的新功能[KYLIN-3795]特此感谢滴滴靳国卫同学对此功能的贡献。Livy 介绍Apache Livy 是一个基于 Spark 的开源 REST 服务是 Apache 基金会的一个孵化项目它能够通过 REST 的方式将代码片段或是序列化的二进制代码提交到 Spark 集群中去执行。它提供了如下基本功能提交 Scala、Python 或是 R 代码片段到远端的 Spark 集群上执行。提交 Java、Scala、Python 所编写的 Spark 作业到远端的 Spark 集群上执行。Apache Livy 架构为什么使用 Livy1. 当前 Spark 存在的问题Spark 当前支持两种交互方式交互式处理用户使用 spark-shell 或 pyspark 脚本启动 Spark 应用程序伴随应用程序启动的同时Spark 会在当前终端启动 REPL(Read–Eval–Print Loop) 来接收用户的代码输入并将其编译成 Spark 作业。批处理批处理的程序逻辑由用户实现并编译打包成 jar 包spark-submit 脚本启动 Spark 应用程序来执行用户所编写的逻辑与交互式处理不同的是批处理程序在执行过程中用户没有与 Spark 进行任何的交互。两种方式都需要用户登录到 Gateway 节点上通过脚本启动 Spark 进程但是会出现以下问题增加 Gateway 节点的资源使用负担和故障发生的可能性。同时 Gateway 节点的故障会带来单点问题造成 Spark 程序的失败。难以管理、审计以及与已有的权限管理工具的集成。由于 Spark 采用脚本的方式启动应用程序因此相比于 WEB 方式少了许多管理、审计的便利性同时也难以与已有的工具结合如 Apache Knox 等。将 Gateway 节点上的部署细节以及配置不可避免地暴露给了登陆用户。2. Livy 优势一方面接受并解析用户的 REST 请求转换成相应的操作另一方面它管理着用户所启动的所有的 Spark 集群。Livy 具有如下功能通过 Livy session 实时提交代码片段与 Spark 的 REPL 进行交互。通过 Livy batch 提交 Scala、Java、Python 编写的二进制包来提交批处理任务。多用户能够使用同一个服务器(支持用户模拟)。能够通过 REST 接口在任何设备上提交任务、查看任务执行状态和结果。Kylin with Livy1. 引入 Livy 之前 Kylin 是如何使用 Spark 的Spark 是在 Kylin v2.0 引入的主要应用于 Cube 构建构建过程介绍可以查看https://kylin.apache.org/blog/2017/02/23/by-layer-spark-cubing/下面是 SparkExecutable 类的 doWork 方法关于提交 Spark job 的一段代码,我们可以看到 Kylin 会从配置中获取 Spark job 包的路径(默认为 $KYLIN_HOME/lib)通过本地指令的形式提交 Spark job然后循环获取 Spark job 的执行状态和结果。我们可以看到 Kylin 单独开了一个线程在本地向 Spark 客户端发送来 job 请求并且循环获取结果额外增加了节点系统压力。Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { //略... String jobJar config.getKylinJobJarPath(); //获取job jar的路径 //略... final String cmd String.format(Locale.ROOT, stringBuilder.toString(), hadoopConf,KylinConfig.getSparkHome(), jars, jobJar, formatArgs()); //构建本地command //略... //创建指令执行线程 Callable callable new CallablePairInteger, String() { Override public PairInteger, String call() throws Exception { PairInteger, String result; try { result exec.execute(cmd, patternedLogger); } catch (Exception e) { logger.error(error run spark job:, e); result new Pair(-1, e.getMessage()); } return result; } }; //略... try { FuturePairInteger, String future executorService.submit(callable); PairInteger, String result null; while (!isDiscarded() !isPaused()) { if (future.isDone()) { result future.get(); //循环获取指令执行结果 break; } else { Thread.sleep(5000); //每隔5秒检查一次job执行状态 } } //略... } catch (Exception e) { logger.error(Error run spark job:, e); return ExecuteResult.createError(e); } //略... }2. Livy for Kylin 详细解析Livy 向 Spark 提交 job 一共有两种分别是 Session 和 BatchKylin 是通过 Batch 的方式提交 job 的需要提前构建好 Spark job 对应的 jar 包并上传到 HDFS 中并且将配置项 kylin.engine.livy-conf.livy-key.filehdfs:///path-to-kylin-job-jar 加入到 kyiln.properties 中。Batch 一共具有如下九种状态public enum LivyStateEnum { starting, running, success, dead, error, not_started, idle, busy, shutting_down; }下面是 SparkExecutableLivy 类的 doWork 方法和 LivyRestExecutor 类的 execute 方法关于提交 Spark job 的一段代码Kylin 通过 livyRestBuilder 读取配置文件获取 Spark job 的包路径然后通过 restClient 向 Livy 发送 Http 请求。在提交 job 之后会每隔 10 秒查询一次 job 执行的结果直到 job 的状态变为 shutting_down, error, dead, success 中的一种。每一次都是通过 Http 的方式发送请求相比较于通过本地 Spark 客户端提交任务更加稳定而且减少了 Kylin 节点系统压力。Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { //略... livyRestBuilder.setLivyTypeEnum(LivyTypeEnum.job); executor.execute(livyRestBuilder, patternedLogger); //调用LivyRestExecutor类的execute方法 if (isDiscarded()) { return new ExecuteResult(ExecuteResult.State.DISCARDED, Discarded); } if (isPaused()) { return new ExecuteResult(ExecuteResult.State.STOPPED, Stopped); } //略... } public void execute(LivyRestBuilder livyRestBuilder, Logger logAppender) { LivyRestClient restClient new LivyRestClient(); String result restClient.livySubmitJobBatches(dataJson); //向Livy发送http请求 JSONObject resultJson new JSONObject(result); String state resultJson.getString(state); //得到Livy请求结果 final String livyTaskId resultJson.getString(id); while (!LivyStateEnum.shutting_down.toString().equalsIgnoreCase(state) !LivyStateEnum.error.toString().equalsIgnoreCase(state) !LivyStateEnum.dead.toString().equalsIgnoreCase(state) !LivyStateEnum.success.toString().equalsIgnoreCase(state)) { String statusResult restClient.livyGetJobStatusBatches(livyTaskId); //获取Spark job执行状态 JSONObject stateJson new JSONObject(statusResult); if (!state.equalsIgnoreCase(stateJson.getString(state))) { logAppender.log(Livy status Result: stateJson.getString(state)); } state stateJson.getString(state); Thread.sleep(10*1000); //每10秒检查一次结果 } }3. Livy 在 Kylin 中的应用构建 Intermediate Flat Hive Table 和 Redistribute Flat Hive Table 原本都是通过 Hive 客户端Cli 或 Beeline进行构建的引入 Livy 之后Kylin 通过 Livy 来调用 SparkSQL 进行构建提高了平表的构建速度。在引入 Livy 之后Cube 的构建主要改变的是以下几个步骤对应的任务日志输出如下构建 Intermediate Flat Hive Table构建 Redistribute Flat Hive Table使用 Spark-Submit 的地方都用 Livy 的 Batch API 进行替换1构建 Cube2转换 Cuboid 为 HFile4. 引入 Livy 对 Kylin 的好处无需准备 Spark 的客户端配置Kylin 部署更加轻量化。Kylin 节点系统压力更低无需在 Kylin 节点启动 Spark 客户端。构建 Flat Hive Table 更快通过 Livy 可以使用 Spark SQL 构建平表而 Spark SQL 要快于 Hive。提交 job 更快job 状态获取更方便。5. 如何在 Kylin 中启用 Livy在 Kylin 启用 Livy 前请先确保 Livy 能够正常工作1在 Kylin.properties 中加入如下配置并重启使之生效。//此处为CDH5.7环境下的配置 kylin.engine.livy-conf.livy-enabledtrue kylin.engine.livy-conf.livy-urlhttp://cdh-client:8998 kylin.engine.livy-conf.livy-key.filehdfs:///path/kylin-job-3.0.0-SNAPSHOT.jar //请根据个人环境替换对应版本的包 kylin.engine.livy-conf.livy-arr.jarshdfs:///path/hbase-client-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-common-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-hadoop-compat-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-hadoop2-compat-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-server-1.2.0-cdh5.7.5.jar,hdfs:///path/htrace-core-3.2.0-incubating.jar,hdfs:///path/metrics-core-2.2.0.jar其中 livy-key.file 和 livy-arr.jars 地址之间不要有空格否则可能会出不可预知的错误。2Cube 构建引擎选用 Spark。常见问题以下问题往往为使用不当和配置错误的原因非 Kylin 本身存在的问题此处仅为友情提示。1. Table or view not found输出日志Exception in thread main org.apache.spark.sql.AnalysisException: Table or view not found: DEFAULT.KYLIN_SALES; line 21 pos 6;解决方法//将hive-site.xml拷贝到spark的配置文件目录中 ln -s /etc/hive/conf/hive-site.xml $SPARK_CONF_DIR2. livy request 400 error解决方法//kylin.properties Livy配置项jar包地址之间不要留空格 //此处为CDH5.7环境下的依赖包请根据个人环境替换对应版本的包 kylin.engine.livy-conf.livy-arr.jarshdfs:///path/hbase-client-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-common-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-hadoop-compat-1.2.0-cdh5.7.5.jar,hdfs:///path/hbase-hadoop2-compat-1.2.0-cdh5.7.5.jar,hdfs:///path/3. NoClassDefFoundError输出日志NoClassDefFoundError: org/apache/hadoop/hbase/protobuf/generated/HFileProtos解决方法find /opt -type f -name hbase-protocol*.jar cp /path/to/hbase-protocol-1.2.0-cdh5.7.5.jar $SPARK_HOME/jars4. livy sql 执行错误解决方法//kylin.properties中添加如下配置 kylin.source.hive.quote-enabledfalse总结Livy 本质上是在 Spark 上的 REST 服务对于 Kylin cube 的构建没有本质上的性能提升但是通过引入 LivyKylin 能够直接通过 Spark SQL 代替 Hive 构建 Flat Table而且管理 Spark job 也更加方便。但是Livy 当前也存在一些问题比如使用较低或较高版本的 Spark 无法正常工作以及单点故障等问题用户可以考虑自身的实际场景选择是否需要在 Kylin 中使用 Livy。参考文章https://hortonworks.com/blog/livy-a-rest-interface-for-apache-spark/https://wiki.apache.org/incubator/LivyProposalhttps://kylin.apache.org/blog/2017/02/23/by-layer-spark-cubing/作者简介王汝鹏Kyligence 大数据研发工程师主要负责 Apache Kylin 社区维护和开发。GitHubhttps://github.com/rupengwang。了解更多大数据资讯点击进入Kyligence官网
