- Apache Kylin with Apache Livy(incubator)
- 前言
- Livy介绍
- 为什么使用Livy
- 当前Spark存在的问题
- Livy优势
- Kylin with Livy
- 引入Livy之前Kyin是如何使用Spark的
- Livy for Kylin详细解析
- Livy在Kylin中的应用
- 引入Livy对kylin的好处
- 如何在Kylin中启用Livy
- 常见问题
- 总结
- 参考文章
一、前言
为了更方便地向Spark提交、管理和监控任务，有些用户会使用Livy作为Spark的交互接口，在最新的Apache Kylin3.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 Live架构
三、为什么使用Livy
1. 当前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 Livy
1. 引入Livy之前Kyin是如何使用Spark的
Spark是在Kylin v2.0引入的，主要应用于Cube构建，构建过程介绍可以查看这篇博客
下面是SparkExecutable类的doWork方法关于提交Spark job的一段代码,我们可以看到Kylin会从配置中获取Spark job包的路径(默认为$KYIN_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 Callable<Pair<Integer, String>>() {
@Override
public Pair<Integer, String> call() throws Exception {
Pair<Integer, 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 {
Future<Pair<Integer, String>> future = executorService.submit(callable);
Pair<Integer, 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和Batch，Kyin是通过Batch的方式提交job的，需要提前构建好Spark job对应的jar包并上传到HDFS中，并且将配置项
kylin.engine.livy-conf.livy-key.file=hdfs:///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进行替换
-
构建Cube
-
转换Cuboid为HFile
-
4. 引入Livy对kylin的好处
- 无需准备Spark的客户端配置，Kylin部署更加轻量化
- Kylin节点系统压力更低，无需在Kylin节点启动Spark客户端
- 构建Flat Hive Table更快，通过Livy可以使用Spark SQL构建平表，而Spark SQL要快于Hive
- 提交job更快，job状态获取更方便
5. 如何在Kylin中启用Livy
在Kylin启用Livy前，请先确保Livy能够正常工作
-
在Kylin.properties中，加入如下配置，并重启使之生效
//此处为CDH5.7环境下的配置 kylin.engine.livy-conf.livy-enabled=true kylin.engine.livy-conf.livy-url=http://cdh-client:8998 kylin.engine.livy-conf.livy-key.file=hdfs:///path/kylin-job-3.0.0-SNAPSHOT.jar //请根据个人环境替换对应版本的包 kylin.engine.livy-conf.livy-arr.jars=hdfs:///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地址之间不要有空格，否则可能会出不可预知的错误。
-
Cube构建引擎选用Spark
五、常见问题
以下问题往往为使用不当和配置错误的原因，非Kylin本身存在的问题，此处仅为友情提示。
-
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_DIR -
livy request 400 error
解决方法：//kylin.properties Livy配置项jar包地址之间不要留空格 //此处为CDH5.7环境下的依赖包，请根据个人环境替换对应版本的包 kylin.engine.livy-conf.livy-arr.jars=hdfs:///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/ -
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/jars -
livy sql执行错误
解决方法：//kylin.properties中添加如下配置 kylin.source.hive.quote-enabled=false
六、总结
Livy本质上是在Spark上的REST服务，对于Kylin cube的构建没有本质上的性能提升，但是通过引入Livy，Kyin能够直接通过Spark SQL代替Hive构建Flat Table，而且管理Spark job也更加方便。但是Livy当前也存在一些问题，比如使用较低或较高版本的Spark无法正常工作以及单点故障等问题，用户可以考虑自身的实际场景选择是否需要在Kylin中使用Livy。
网友评论