您的位置:首页 > 其它

Spark HiveThriftServer2启动流程源码分析

2017-05-06 17:30 871 查看

背景

接触SparkSQL不久,查找了些别人的资料,感觉对整个Spark HiveThriftServer2流程讲的糊里糊涂的,觉得需要从Beeline连接HiveThriftServer2开始,梳理下执行SQL的流程。

由于公司Spark集群对上层提供的主要是SparkSQL服务,之前对Hive2也接触较少,总好奇运行在Spark On Yarn的任务,是如何同Hive2做交互的,最终真正的物理计划(Task)又是如何跑在Spark上。

该文比较简单地梳理下,Spark HiveThrift2是如何重载HiveCli/HiveSession/OperationManger等实现从Beeline入口,到Spark SQL上的操作的。

随后的SQL解析、逻辑计划、物理计划、Spark On Yarn的执行,找时间一连串的梳理一遍。

一、 Spark HiveThriftServer2启动流程

1. 从Beeline开始

Beeline代码没有认真对,主要是找出Beeline->HiveCli交互的入口,实际上Beeline是通过JDBC同HiveCli进行连接的。

beeline.main()->mainWithInputRedirection()->begin()->execute()->dispatch()

boolean dispath(String line){
...
if (isBeeLine) {
if (line.startsWith(COMMAND_PREFIX)) {
// handle SQLLine command in beeline which starts with ! and does not end with ;
return execCommandWithPrefix(line);
} else {
return commands.sql(line, getOpts().getEntireLineAsCommand());
}
} else {
return commands.sql(line, getOpts().getEntireLineAsCommand());
}
}


Commands.sql()->execute()->executeInternal()

private boolean executeInternal(String sql, boolean call) {

...

try {
Statement stmnt = null;
boolean hasResults;
Thread logThread = null;

try {
long start = System.currentTimeMillis();

if (call) {
stmnt = beeLine.getDatabaseConnection().getConnection().prepareCall(sql);
hasResults = ((CallableStatement) stmnt).execute();
} else {
// 创建statement(HiveStatement重载了sql.Statement)
stmnt = beeLine.createStatement();

if (beeLine.getOpts().isSilent()) {
// 执行sql
hasResults = stmnt.execute(sql);
} else {
logThread = new Thread(createLogRunnable(stmnt));
logThread.setDaemon(true);
logThread.start();
hasResults = stmnt.execute(sql);
logThread.interrupt();
logThread.join(DEFAULT_QUERY_PROGRESS_THREAD_TIMEOUT);
}
}
}finally{
...
}
}


HiveStatement真正处理sql的方法:

public boolean execute(String sql) throws SQLException {
// 此处为真正处理sql的函数
runAsyncOnServer(sql);

waitForOperationToComplete();

// 此处为已经查询完毕,format输出结果
// The query should be completed by now
if (!stmtHandle.isHasResultSet()) {
return false;
}
resultSet =  new HiveQueryResultSet.Builder(this).setClient(client).setSessionHandle(sessHandle)
.setStmtHandle(stmtHandle).setMaxRows(maxRows).setFetchSize(fetchSize)
.setScrollable(isScrollableResultset)
.build();
return true;
}

// 异步地同Server交互(其实是Cli层)
private void runAsyncOnServer(String sql) throws SQLException {
checkConnection("execute");

closeClientOperation();
initFlags();

TExecuteStatementReq execReq = new TExecuteStatementReq(sessHandle, sql);
/**
* Run asynchronously whenever possible
* Currently only a SQLOperation can be run asynchronously,
* in a background operation thread
* Compilation is synchronous and execution is asynchronous
*/
execReq.setRunAsync(true);
execReq.setConfOverlay(sessConf);
execReq.setQueryTimeout(queryTimeout);
try {
// client 为TCLIService.Iface类型,基于RPC执行该Statement

TExecuteStatementResp execResp = client.ExecuteStatement(execReq);
Utils.verifySuccessWithInfo(execResp.getStatus());
stmtHandle = execResp.getOperationHandle();
isExecuteStatementFailed = false;
} catch (SQLException eS) {
isExecuteStatementFailed = true;
throw eS;
} catch (Exception ex) {
isExecuteStatementFailed = true;
throw new SQLException(ex.toString(), "08S01", ex);
}
}

public abstract class ThriftCLIService extends AbstractService implements TCLIService.Iface, Runnable {
...
}


2. 走到了ThriftCLIService

先来看看被beeline调用的ExecuteStatement()函数做了什么事情,其就是基于Thrift协议获取请求信息,同时将cliService执行的结果以Thrift协议返回。真正的执行还要往下走。

@Override
public TExecuteStatementResp ExecuteStatement(TExecuteStatementReq req) throws TException {
TExecuteStatementResp resp = new TExecuteStatementResp();
try {
SessionHandle sessionHandle = new SessionHandle(req.getSessionHandle());
String statement = req.getStatement();
Map<String, String> confOverlay = req.getConfOverlay();
Boolean runAsync = req.isRunAsync();
long queryTimeout = req.getQueryTimeout();

//此处cliService为CliService变量,真正执行操作还要往下走(快要逼近真相了..)
OperationHandle operationHandle =
runAsync ? cliService.executeStatementAsync(sessionHandle, statement, confOverlay,
queryTimeout) : cliService.executeStatement(sessionHandle, statement, confOverlay,
queryTimeout);
resp.setOperationHandle(operationHandle.toTOperationHandle());
resp.setStatus(OK_STATUS);
} catch (Exception e) {
// Note: it's rather important that this (and other methods) catch Exception, not Throwable;
// in combination with HiveSessionProxy.invoke code, perhaps unintentionally, it used
// to also catch all errors; and now it allows OOMs only to propagate.
LOG.warn("Error executing statement: ", e);
resp.setStatus(HiveSQLException.toTStatus(e));
}
return resp;
}


3. CliService做了什么?

此处为破解Spark同HiveCli交互真相的关键,看到之类所有的方法类,是不是都在Spark/sql/HiveThrift下见过相似的变量。

真相其实很简单,Spark HiveThrift2所做的是事情就是将下面这个处理用到的SessionManager/OperationManager/CliService方法重载下,下面这个方法调用的时候就跑到SparkSQL上了。

不信,你往下看。

/**
* Execute statement on the server with a timeout. This is a blocking call.
*/
@Override
public OperationHandle executeStatement(SessionHandle sessionHandle, String statement,
Map<String, String> confOverlay, long queryTimeout) throws HiveSQLException {
OperationHandle opHandle =
sessionManager.getSession(sessionHandle).executeStatement(statement, confOverlay,
queryTimeout);
LOG.debug(sessionHandle + ": executeStatement()");
return opHandle;
}


4. 回到SparkSQLCliService

SparkSQLCLIService继承CLIService,同时重载了init()方法。init()方法,将sessionManager至为SparkSQLSessionManager()类。

这里有趣的ReflectedCompositeService类,基于反射原理设置制定父类的变量类型。(不细分析了,原理有别的文章解释)

private[hive] class SparkSQLCLIService(
hiveServer: HiveServer2)
extends CLIService(hiveServer)
with ReflectedCompositeService {

override def init(hiveConf: HiveConf) {
this.hiveConf = hiveConf
this.sessionManager = new SparkSQLSessionManager(hiveServer)
addService(sessionManager)
this.serviceUGI = MultiSparkSQLEnv.globalUgi
initCompositeService(hiveConf)
}

...

private[thriftserver] trait ReflectedCompositeService { this: AbstractService =>
def initCompositeService(hiveConf: HiveConf) {
// Emulating `CompositeService.init(hiveConf)`
val serviceList = getAncestorField[JList[Service]](this, 2, "serviceList")
serviceList.asScala.foreach(_.init(hiveConf))

// Emulating `AbstractService.init(hiveConf)`
invoke(classOf[AbstractService], this, "ensureCurrentState", classOf[STATE] -> STATE.NOTINITED)
setAncestorField(this, 3, "hiveConf", hiveConf)
invoke(classOf[AbstractService], this, "changeState", classOf[STATE] -> STATE.INITED)
getAncestorField[Log](this, 3, "LOG").info(s"Service: $getName is inited.")
}

}


5. SparkSQLSessionManager做了些什么?

SparkSQLSessionManager为理解这一环最绕的一个地方,它通过重载openSession()来实现SparkSQLOperationManager的调用。

sessionManager.getSession(sessionHandle).executeStatement()

SparkSQLSessionManager类的具体实现:

private[hive] class SparkSQLSessionManager(hiveServer: HiveServer2, sqlContext: SQLContext)
extends SessionManager(hiveServer)
with ReflectedCompositeService {

// !!!
private lazy val sparkSqlOperationManager = new SparkSQLOperationManager()

//重载init操作,实现backgroundPool,此为操作operation的线程池
override def init(hiveConf: HiveConf) {
setSuperField(this, "hiveConf", hiveConf)

// Create operation log root directory, if operation logging is enabled
if (hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_LOGGING_OPERATION_ENABLED)) {
invoke(classOf[SessionManager], this, "initOperationLogRootDir")
}

val backgroundPoolSize = hiveConf.getIntVar(ConfVars.HIVE_SERVER2_ASYNC_EXEC_THREADS)
setSuperField(this, "backgroundOperationPool", Executors.newFixedThreadPool(backgroundPoolSize))
getAncestorField[Log](this, 3, "LOG").info(
s"HiveServer2: Async execution pool size $backgroundPoolSize")

// 此处将父类的operationManager致为sparkSqlOperationManager,在sparkSqlOperationManager类中,实现SparkSQL操作,而非Hive的MapReduce操作
setSuperField(this, "operationManager", sparkSqlOperationManager)
addService(sparkSqlOperationManager)

initCompositeService(hiveConf)
}

override def openSession(
protocol: TProtocolVersion,
username: String,
passwd: String,
ipAddress: String,
sessionConf: java.util.Map[String, String],
withImpersonation: Boolean,
delegationToken: String): SessionHandle = {
// 此处关键:先调用父类的该方法(该方法操作在下面),其实际上将
val sessionHandle =
super.openSession(protocol, username, passwd, ipAddress, sessionConf, withImpersonation,
delegationToken)
val session = super.getSession(sessionHandle)
HiveThriftServer2.listener.onSessionCreated(
session.getIpAddress, sessionHandle.getSessionId.toString, session.getUsername)
val ctx = if (sqlContext.conf.hiveThriftServerSingleSession) {
sqlContext
} else {
sqlContext.newSession()
}
ctx.setConf("spark.sql.hive.version", HiveUtils.hiveExecutionVersion)
if (sessionConf != null && sessionConf.containsKey("use:database")) {
ctx.sql(s"use ${sessionConf.get("use:database")}")
}
sparkSqlOperationManager.sessionToContexts.put(sessionHandle, ctx)
sessionHandle
}
}


Hive SessionManager中的openSession(),实际上是调用createSession()函数完成的,同时创建一个HiveSessionImpl/HiveSessionImplWithUGI/HiveSessionProxy返回。

public HiveSession createSession(SessionHandle sessionHandle, TProtocolVersion protocol, String username,
String password, String ipAddress, Map<String, String> sessionConf, boolean withImpersonation,
String delegationToken)
throws HiveSQLException {

HiveSession session;
// If doAs is set to true for HiveServer2, we will create a proxy object for the session impl.
// Within the proxy object, we wrap the method call in a UserGroupInformation#doAs

if (withImpersonation) {
HiveSessionImplwithUGI hiveSessionUgi;
if (sessionImplWithUGIclassName == null) {
hiveSessionUgi = new HiveSessionImplwithUGI(sessionHandle, protocol, username, password,
hiveConf, ipAddress, delegationToken);
} else {
try {
Class<?> clazz = Class.forName(sessionImplWithUGIclassName);

Constructor<?> constructor = clazz.getConstructor(SessionHandle.class, TProtocolVersion.class, String.class,
String.class, HiveConf.class, String.class, String.class);
hiveSessionUgi = (HiveSessionImplwithUGI) constructor.newInstance(sessionHandle,
protocol, username, password, hiveConf, ipAddress, delegationToken);
} catch (Exception e) {
throw new HiveSQLException("Cannot initilize session class:" + sessionImplWithUGIclassName);
}
}
session = HiveSessionProxy.getProxy(hiveSessionUgi, hiveSessionUgi.getSessionUgi());
hiveSessionUgi.setProxySession(session);
} else {
if (sessionImplclassName == null) {
session = new HiveSessionImpl(sessionHandle, protocol, username, password, hiveConf,
ipAddress);
} else {
try {
Class<?> clazz = Class.forName(sessionImplclassName);
Constructor<?> constructor = clazz.getConstructor(SessionHandle.class, TProtocolVersion.class,
String.class, String.class, HiveConf.class, String.class);

// 实际上是调用Hive中的HiveSessionImpl/HiveSessionImplWithUGI/HiveSessionProxy类... 实现的

session = (HiveSession) constructor.newInstance(sessionHandle, protocol, username, password,
hiveConf, ipAddress);
} catch (Exception e) {
throw new HiveSQLException("Cannot initilize session class:" + sessionImplclassName, e);
}
}
}
// 此处是将sparkSqlOperationManager赋值于此,在HiveSessionImpl中会使用
session.setSessionManager(this);
session.setOperationManager(operationManager);
try {
session.open(sessionConf);
} catch (Exception e) {
LOG.warn("Failed to open session", e);
try {
session.close();
} catch (Throwable t) {
LOG.warn("Error closing session", t);
}
session = null;
throw new HiveSQLException("Failed to open new session: " + e.getMessage(), e);
}
if (isOperationLogEnabled) {
session.setOperationLogSessionDir(operationLogRootDir);
}
try {
executeSessionHooks(session);
} catch (Exception e) {
LOG.warn("Failed to execute session hooks", e);
try {
session.close();
} catch (Throwable t) {
LOG.warn("Error closing session", t);
}
session = null;
throw new HiveSQLException("Failed to execute session hooks: " + e.getMessage(), e);
}
handleToSession.put(session.getSessionHandle(), session);
LOG.info("Session opened, " + session.getSessionHandle() + ", current sessions:" + getOpenSessionCount());
return session;
}


既然getSession()方法返回的
HiveSessionImpl/HiveSessionImplWithUGI/HiveSessionProxy
, 那么自然而然都要实现executeStatement()方法了。而在这些类的内部,真正完成操作的是executeStatementInternal()函数,该函数实现很简单,具体如下:

private OperationHandle executeStatementInternal(String statement,
Map<String, String> confOverlay, boolean runAsync, long queryTimeout) throws HiveSQLException {
acquire(true, true);

ExecuteStatementOperation operation = null;
OperationHandle opHandle = null;
try {

// 看到了没? 由于之前我们重置了operationManager变量,此处实际上调用的是sparkSqlOperationManager类,总算解开谜底
operation = getOperationManager().newExecuteStatementOperation(getSession(), statement,
confOverlay, runAsync, queryTimeout);
opHandle = operation.getHandle();
operation.run();
addOpHandle(opHandle);
return opHandle;
} catch (HiveSQLException e) {
// Refering to SQLOperation.java, there is no chance that a HiveSQLException throws and the
// async background operation submits to thread pool successfully at the same time. So, Cleanup
// opHandle directly when got HiveSQLException
if (opHandle != null) {
getOperationManager().closeOperation(opHandle);
}
throw e;
} finally {
if (operation == null || operation.getBackgroundHandle() == null) {
release(true, true); // Not async, or wasn't submitted for some reason (failure, etc.)
} else {
releaseBeforeOpLock(true); // Release, but keep the lock (if present).
}
}
}


Operation类为一个抽象类,继承该类的对象需要实现run()/internalRun()方法,在该执行操作中,真正执行为operator.run()。

6. SparkSQLOperationManager类

该类重载了newExecuteStatementOperation()函数, 最终返回的已经是一个SparkExecuteStatementOperation()对象了。

override def newExecuteStatementOperation(
parentSession: HiveSession,
statement: String,
confOverlay: JMap[String, String],
async: Boolean): ExecuteStatementOperation = synchronized {

val sessionHandle = parentSession.getSessionHandle
val sparkSession = sessionToSparkSession.get(sessionHandle)
var client = sessionToClient.get(sessionHandle)
val formatted = statement.toLowerCase.split("//s+").mkString(" ")
...

val sessionState = sparkSession.sessionState.asInstanceOf[HiveSessionState]
val runInBackground = async && sessionState.hiveThriftServerAsync
val operation = new SparkExecuteStatementOperation(
parentSession,
statement,
client,
confOverlay,
runInBackground)(sparkSession, sessionToActivePool)
handleToOperation.put(operation.getHandle, operation)
logDebug(s"Created Operation for $statement with session=$parentSession, " +
s"runInBackground=$runInBackground")
operation
}


7. 一条SQL真正的命运

上文生成的SparkExecuteStatementOperation对象,调用runInternal()方法调用的是execute()。execute()真正调用的则是
result = sparkSession.sql(statement)
, 做了这么多准备,真正的实现其实就像我们自己写个demo时的ss.sql()操作,感觉好费周折啊。

private def execute(): Unit = {
statementId = UUID.randomUUID().toString
logInfo(s"Running query '$statement' with $statementId")
setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sparkSession.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
statement,
statementId,
parentSession.getUsername)
sparkSession.sparkContext.setJobGroup(statementId, statement)
val pool = sessionToActivePool.get(parentSession.getSessionHandle)
if (pool != null) {
sparkSession.sparkContext.setLocalProperty("spark.scheduler.pool", pool)
}
try {
// 一条SQL真正的命运
result = sparkSession.sql(statement)
logDebug(result.queryExecution.toString())
result.queryExecution.logical match {
case SetCommand(Some((SQLConf.THRIFTSERVER_POOL.key, Some(value)))) =>
sessionToActivePool.put(parentSession.getSessionHandle, value)
logInfo(s"Setting spark.scheduler.pool=$value for future statements in this session.")
case _ =>
}
HiveThriftServer2.listener.onStatementParsed(statementId, result.queryExecution.toString())
iter = {
val useIncrementalCollect =
sparkSession.conf.get("spark.sql.thriftServer.incrementalCollect", "false").toBoolean
if (useIncrementalCollect) {
result.toLocalIterator.asScala
} else {
result.collect().iterator
}
}
val (itra, itrb) = iter.duplicate
iterHeader = itra
iter = itrb
dataTypes = result.queryExecution.analyzed.output.map(_.dataType).toArray
} catch {
case e: HiveSQLException =>
if (getStatus().getState() == OperationState.CANCELED) {
return
} else {
setState(OperationState.ERROR)
throw e
}
// Actually do need to catch Throwable as some failures don't inherit from Exception and
// HiveServer will silently swallow them.
case e: Throwable =>
val currentState = getStatus().getState()
logError(s"Error executing query, currentState $currentState, ", e)
setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw new HiveSQLException(e.toString)
}
setState(OperationState.FINISHED)
HiveThriftServer2.listener.onStatementFinish(statementId)
}


二、总结

本文没有细枝末节的分析各个方法和变量,只是在脑海中走一遍一条SQL在Spark HiveThriftServer2环境中的运行流程,看看大致如何处理的。

而深有感触的是,基于Spark借Hive的壳子,通过java/scala语言的重载功能,很简单、方便的将原Hive功能扩展到Spark上,非常巧妙。

同时基于JDBC查询SQL的功能,中间经过了HiveCommands/HiveStatement/ThriftCli/CliService/SessionManager/SessionImpl/OperationManager/Operator等等许多层的抽象,虽然体会起来很繁杂,但之于上述所说的扩展工作,却显得万分必要。后续工作中,切记学习该项工作。

当然,此文仅仅分析了一条SQL刚刚开始执行的非常简单的逻辑,后续会着手再往下走,一条SQL是如何转变成一个个基于executor执行的task的,把Spark的逻辑梳理一遍。

参考

http://www.datastart.cn/tech/2016/06/01/spark-sql-1.html
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签:  spark hive yarn 源码 thrift