一、在spark中查看执行完成的日志
spark thrift server的web ui在运行时可以看到sql查询的提交用户,执行sql等信息
但是当这个实例停掉或者异常终止以后,你再去spark history server的webui去查看,发现这部分信息就没有了……
image.png
究其原因,原来spark thrift server并没有将这部分数据序列化到spark history server的store中,我们可以构建一个history服务来查看历史日志。回头有空可以单独讲讲这部分源码的实现
这篇帖子是使用一个折中的办法实现了这部分数据的日志留存
修改spark-hive-thriftserver工程下org.apache.spark.sql.hive.thriftserver.HiveThriftServer2类,做如下修改:
def onStatementError(id: String, errorMessage: String, errorTrace: String): Unit = {
synchronized {
executionList(id).finishTimestamp = System.currentTimeMillis
executionList(id).detail = errorMessage
executionList(id).state = ExecutionState.FAILED
totalRunning -= 1
//增加下面一句话
SqlListenerUtil.write(executionList(id))
trimExecutionIfNecessary()
}
}
def onStatementFinish(id: String): Unit = synchronized {
executionList(id).finishTimestamp = System.currentTimeMillis
executionList(id).state = ExecutionState.FINISHED
totalRunning -= 1
//增加下面一句话
SqlListenerUtil.write(executionList(id))
trimExecutionIfNecessary()
}
新增org.apache.spark.sql.hive.thriftserver.SqlListenerUtil类
package org.apache.spark.sql.hive.thriftserver
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import org.apache.spark.internal.Logging
import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{ExecutionInfo, uiTab}
import org.apache.spark.status.api.v1.{JobData, StageData}
import scala.collection.mutable.ArrayBuffer
object SqlListenerUtil extends Logging {
val mapper: ObjectMapper with ScalaObjectMapper = new ObjectMapper() with ScalaObjectMapper
mapper.registerModule(DefaultScalaModule)
val stagesInfo: ArrayBuffer[StageData] = ArrayBuffer[StageData]()
val jobsInfo: ArrayBuffer[JobData] = ArrayBuffer[JobData]()
def write(executionInfo: ExecutionInfo) = synchronized {
stagesInfo.clear()
jobsInfo.clear()
val sparkUI = uiTab.get.parent
val store = sparkUI.store
executionInfo.jobId.foreach {
id =>
val jobData = store.job(id.toInt)
jobsInfo += jobData
jobData.stageIds.foreach {
stageId =>
val stageDatas = store.stageData(stageId)
stagesInfo ++= stageDatas
}
}
val sqlInfo = SqlInfo(sparkUI.appId, executionInfo, jobsInfo, stagesInfo)
log.info(mapper.writeValueAsString(sqlInfo))
}
case class SqlInfo(appId: String, executionInfo: ExecutionInfo, jobsInfo: ArrayBuffer[JobData], stagesInfo: ArrayBuffer[StageData])
}
重新打包编辑后替换相应的jar包
修改spark安装目录下的log4j.properties,增加如下信息:
# 自定义sql查询监控
log4j.logger.org.apache.spark.sql.hive.thriftserver.SqlListenerUtil=INFO,listener
log4j.additivity.org.apache.spark.sql.hive.thriftserver.SqlListenerUtil=false
log4j.appender.listener=org.apache.log4j.DailyRollingFileAppender
log4j.appender.listener.File=/var/log/spark2/spark-sql-listener
log4j.appender.listener.layout=org.apache.log4j.PatternLayout
log4j.appender.listener.layout.ConversionPattern=%m%n
log4j.appender.listener.DatePattern=.yyyy-MM-dd
重启spark-thrift-server
这样查询日志就以json格式记录在/var/log/spark2/spark-sql-listener文件中了
二、返回执行进度日志
问题背景
当前Spark Thrift Server在运行SQL时,Client不会输出任何日志,只有任务运行结束后输出的查询结果集(目前Spark最新版本2.4.0仍有此问题)。在任务运行过程中看不到当前的执行进度,影响了用户的使用体验,为了解决此问题对Spark Thrift Server作了一番改造。
此前为了解决任务提交入口收敛的问题也对HiveServer2作了改造,了解过其内部的相关具体实现,而Spark Thrift Server的核心代码移植自HiveServer2,因此想到通过对比两者的实现差异找到解决问题的切入点。
相关实现
Hive的查询结果和执行日志都是通过fetchResults 方法返回,由 FetchType 参数指定返回的是查询结果(QUERY_OUTPUT)还是执行日志(LOG)。
任务的执行日志在具体代码实现中被称为OperationLog,有关日志的读、写、删除操作都封装在该类中。OperationLog以本地文件的方式存储在 hive.server2.logging.operation.log.location 配置的目录中,每个session有唯一的UUID,对应一个存储目录,相同session内提交的statement用单独的文件存储日志(以提交这个statement的handle UUID作为对应的文件名)。
执行日志的写入并非直接调用OperationLog.writeOperationLog方法,而是专门实现了一个Log4j Appender,通过动态追加Appender,其他类在调用log.info()等方法打印日志时就会统一写入OperationLog。由于在调用log.info()时并不会传入与OperationLog有关的参数,为了区分哪些日志应该写入到哪个OperationLog中,OperationLog内部使用了ThreadLocal与线程做了绑定,也就是说在相同线程上调用log.info打印的日志将写入到相同线程的OperationLog中。
通过对比发现,Spark Thrift Server不输出日志是由于在执行前没有将OperationLog与线程绑定,涉及改动代码:
重新编译,然后测试,发现也就只有寥寥几行日志,依然没有进度信息:
但Thrift Server端却有较详细的进度日志,包括各个stage、task的运行情况等,由此推断 TaskSetManager 的方法调用与OperationLog不在相同线程内。
为了获取各个stage、task的运行信息,可以通过注册SparkListener 实现。
首先在 SparkExecuteStatementOperation 中保存statementId 与 OperationLog 实例的映射关系,然后在SparkListener 通过statementId 得到对用的OperationLog 实例实现执行日志的写入。另外,由于SparkListener中的各个接口参数信息不全的关系,在SparkListener 内部还需要额外存储 jobId 与statementId、stageId 与jobId 的映射关系以辅助找到对应的OperationLog 实例。
最后一步,在 HiveThriftServer2 中注册该SparkListener:
最终输出执行日志效果: