SparkSQL邏輯計(jì)劃原理

【導(dǎo)讀】本文重點(diǎn)講解了SparkSQL解析為AST抽象語(yǔ)法樹、生成

UnresolvedLogicalPlan>生成ResolvedLogicalPlan以及Optimized

LogicalPlan的過程，為接下來進(jìn)一步生成物理計(jì)劃SparkPlan做好了準(zhǔn)各。

Catalyst優(yōu)化器是Spark引擎中非常重要的組成部分，也是近年來Spark社區(qū)

項(xiàng)目重點(diǎn)投入、并且發(fā)展十分迅速的核心模塊，對(duì)于Spark任務(wù)的性能提升起

到了關(guān)鍵的基礎(chǔ)作用。

我們知道，在Spark1.6之前開發(fā)人員是通過Spark的RDD編程接口來實(shí)現(xiàn)對(duì)大

規(guī)模數(shù)據(jù)的分析和處理的，到了Sparkl.6版本后推出了DataSetDataFrame

的編程接口，這種數(shù)據(jù)結(jié)構(gòu)與RDD的主要區(qū)別在于其攜帶了結(jié)構(gòu)化數(shù)據(jù)的

Schema信息，從而可以被SparkCatalyst用來做進(jìn)一步的解析和優(yōu)化；而

SparkSQL則是比DataSet和DataFrame編程接口更為簡(jiǎn)單易用的大數(shù)據(jù)領(lǐng)域

語(yǔ)言，其用戶可以是開發(fā)工程師、數(shù)據(jù)科學(xué)家、數(shù)據(jù)分析師等，并且與其他

SQL語(yǔ)言類似，可以通過SQL引擎將SQL預(yù)先解析成一棵AST抽象語(yǔ)法樹；同

時(shí)，AST抽象語(yǔ)法樹、DalaSet及DalaFrame接下來均會(huì)被SparkCatalyst優(yōu)

化器轉(zhuǎn)換成為UnresolvedLogicalPlansResolvedLogicalPlan,Physical

Plan、以及OptimizedPhysicalPlan,也就是說帶有schema信息的Spark分

布式數(shù)據(jù)集都可以從SparkCatalyst中受益，這也是Spark任務(wù)性能得以提升

的核心所在。

值得一提的是，在物理計(jì)劃樹的生成過程中，首先會(huì)將數(shù)據(jù)源解析成為RDD,

也即在SparkSQL的物理計(jì)劃執(zhí)行過程中所操作的對(duì)象實(shí)際是RDD,一條Spark

SQL在生成最終的物理計(jì)劃后仍然會(huì)經(jīng)過前面文查中所提到的生成DAG、劃分

Stage,并將taskset分發(fā)到特定的executor上運(yùn)行等一系列的任務(wù)調(diào)度和執(zhí)

行過程來實(shí)現(xiàn)該SparkSQL的處理邏輯.

接下來，本文將著重講解SparkSQL邏輯計(jì)劃的相關(guān)實(shí)現(xiàn)原理，在后續(xù)的文章

中會(huì)繼續(xù)解析SparkSQL的物理計(jì)劃。

生成UnresolvedLogicalPlan

用戶可以通過spark-sql等客戶端來提交sql語(yǔ)句，在sparksession初始化時(shí)

通過BaseSessionStateBuilder的build()方法始化SparkSqlParser、

Analyser以及SparkOptimizcr對(duì)象等:

efbuild()：SessionState=

newSessionState

session.sharedState,

conf,

experimentalMethods,

functionRegistry^^^^^^^^M

udfRcgistratiorh^^^^^^^^H

0=>

()=>ana1yzer,

()=>opl'inizor,

planner,

()=>streamingQueryManager,

1istenerManagcr,

()二)resourceLoader,^^^B

createQueryExecutiont

createClone,

columnarRules,

queryStagePrepRules)^^^^^M

當(dāng)用戶程序調(diào)用SparkSession的sql接口時(shí)即開始了解析sql語(yǔ)句并執(zhí)行對(duì)數(shù)

據(jù)處理的過程：

|defsql(sqlText：String)：DataFrame二withActive

|valtracker二newQueryPlanningTracker^^^^^^^^^^^^^^l

valplan=tracker.measurePhase(QucryPlanningTrackcr.PARSING)

sessionState^sqlParser.parsePlan(sqlText)

Dataset.ofRows(self,plan,tracker)

其+1通過AbstractSqlParser的parsePlan方法將sql語(yǔ)句轉(zhuǎn)換成抽象語(yǔ)法

樹：

verridedefparsePlan(sqlText：String)：LogicalPlan=parse(sq

IText){parser二

dslBuilder.visiiSiugleSldleiiienl(paxser.siugleSldLeinenl())nidlcl

caseplan：LogicalPlan=>plan

case

valposition=Origin(None,None)

thrownewParseException(Option(sqlText),"Unsupported

SQLstatement",position,position)

1、從SqlBaseParser的singleStatement()方法開始基于ANTLR4lib庫(kù)來解

析sql語(yǔ)句中所有的詞法片段，生成一棵AST抽象語(yǔ)法樹;

2、訪問AST抽象語(yǔ)法樹并生成Unresolved邏輯計(jì)劃樹:

1)訪問SingleStatementContext節(jié)點(diǎn):

SingleStatementContext是整個(gè)抽象語(yǔ)法樹的根節(jié)點(diǎn)，因此以AstBuilder的

visitSingleStatement方法為入口來遍歷抽象語(yǔ)法樹:

verridedefvisitSingleStatement(ctx：SingleStatementContext)：I

visit(ctx.statcncnt).asInstanccOf[LogicalPlan]

■

)ublicTvisit(ParseTreetree)

returntree.accept(this)：

2）根據(jù)訪問者模式執(zhí)行SingleStatementContext節(jié)點(diǎn)的accept方法:

Override

>ublic<T>Taccept(ParseTreeVisitor<?extendsT>visitor)

if(visitorinstanceofSqlBaseVisitor)return((SqlBas

Wisitor<?extendsT>)visitor).visitSingleStatement(this)；

■

erridcpublicvisitSingleStatementISqlBaseParser.SingleStatcr

ntContextctx){returnvisitChildren(ctx)；}

3）迭代遍歷整棵ASTTree：

Override

intn二node.getChildCount()

(!shouldVisitNextChiId(node,result))

ParseTreec二node.getChild：i)

TchildResult二c.accept(lh：s)

result=aggregateResult(result,chiIdResult)；

returnresult；

根據(jù)以上代碼，在遍歷AST樹的過程中，會(huì)首先解析父節(jié)點(diǎn)的所有子節(jié)點(diǎn)，并

執(zhí)行子節(jié)點(diǎn)上的accept方法來進(jìn)行解析，當(dāng)所有子節(jié)點(diǎn)均解析為

UnresolvedRelation或者Expression后,將這些結(jié)果進(jìn)行聚合并返回到父節(jié)

點(diǎn)，由此可見，AST樹的遍歷所采用的是后序遍歷模式。

接下來以查詢語(yǔ)句中的QuerySpccificationContext節(jié)點(diǎn)的解析為例進(jìn)一步闡

述以上過程:

如下為一條基本的sql語(yǔ)句:

electcol1fromtabnamewhereco!2>1(

QuerySpecificationContext節(jié)點(diǎn)下會(huì)產(chǎn)生用于掃描數(shù)據(jù)源的

FromClauseContext過濾條件對(duì)應(yīng)的BooleanDefaultContexts以及投影時(shí)所

需的NamcdExprcssionScqContext節(jié)點(diǎn)。

1)FromClauseContext繼續(xù)訪問其子節(jié)點(diǎn)，當(dāng)訪問到TablelNameContext節(jié)點(diǎn)

時(shí)，訪問到tableName的tocken時(shí)根據(jù)表名生成UnresolvedRelation：

overridedefvisitTableName(ctx：TableNameContext)：LogicalPlan|

二withOrigin(ctx)

valtableld二visitMultipartldentif：er(ctx.multipartldcntifier|

valtable二nayApplyAliasPlan(ctx.tableAlias,UnresolvedRelat|

2)BooleanDefaultContext的子節(jié)點(diǎn)中分為三個(gè)分支：代表Reference的

ValueExpressionDefaultContext>代表數(shù)值的

ValueExpressionDefaultContext以及代表運(yùn)算符的ComparisonContext；

例如遍歷代表數(shù)據(jù)值ValueExpressionDefaultContext及其子節(jié)點(diǎn)，直到訪問

至ijIntcgerLiteralContext：

verridedefvisitlntegerLiteral(ctx：IntegerLiteralContext)：Lit|

ral二withOrigin：ctx)

BigDccimal(ctx.gctTcxt)match

casevifv.isValidTnt二

Literal：v.intValue)

casevifv.isValidLong二〉

Literal；v.

casev=>Literal(v.underlying(J)

而Literal的定義如下，是一個(gè)葉子類型的Expression節(jié)點(diǎn):

aseclassLiteral(value：Any,dataType：DataType)extends1

cafExpression

3)NamedExpressionSeqContext是投影節(jié)點(diǎn)，迭代遍歷直到

Regu1arQuerySpecificationContext節(jié)點(diǎn)，然后通過訪問

withSelectQuerySpecification方法創(chuàng)建出投影所需的ProjectLogical

Plan：

verridedefvisitRegularQuerySpecification

ctx：RegularQuerySpecificationContext)：LogicalPlan=wit|

Origin(ctx)

valfromOneRowRelation().optiona'(ctx.fromClause)

visitFromClause(ctx.fromClause)

withSclectQuerySpecification

ctx.lateralView,

ctx.\vhereClause,^^^^B

ctx.aggregationsause，?

ctx.havingC：ause,^^^^|

efcreateProjeclC二if(namedExpressions.nonEmply)

Project(namedExpressions,withFi

總結(jié)一下以上處理過程中所涉及的類之間的關(guān)系，如下圖所示:

類圖

生成ResolvedLogicalPlan

SparkAnalyser

在SparkSession的sql方法中，對(duì)sql語(yǔ)句進(jìn)行過Parser解析并生成

UnresolvedLogicalPlan之后則通過執(zhí)行Dataset.ofRows(self,plan,

tracker)繼續(xù)進(jìn)行catalog綁定，數(shù)據(jù)源綁定的過程如下：

|defofRows（sparkSession：SparkSession,logicalPlan：LogicalPlan,trackt

卜：QueryP1anningTracker）

|:DataFrame二sparkSession.wiIhAclive

|\，1q（'二n「v;Query「x（culi（）n（spnrkScssi（）n,I（）gica]PIan,li7i（kr）

,il-l,-.1■,'：：■■/,：?'.I；IJ,q,■.II,I-.....-I.'II：

■

defassertAnalyzedO：Unit=analyzed

由如下實(shí)現(xiàn)邏輯可見，analyzed變量是通過懶加載方式初始化的，通過該變

量的初始方法可見Spark的catalog實(shí)現(xiàn)邏輯主要通過Analyser類來實(shí)現(xiàn)的:

lazyvalanalyzed：LogicalPlan=executePhase(QueryPlanningTracker.ANAI

YS1S)

sparkSession.sessionState.analyzer.executeAndCheck(logical,tracker)

其中,executeAndCheck方法的執(zhí)行是通過Analyzer的父類RuleExecutor的

execute方法來實(shí)現(xiàn)的：

efexecute(plan：TreeType)：TreeType=

■

valbatchStartPlan=curPlan|

variteration二

varcontinue=true

curPlan=batch,rules.foldLeft(curPlan)

case(plan,rule)=.

valstartTime二System.nanoTime

valresult二rule(plan)

valrunTime=System.nanoTi【ne()-startTime|

va1effectivo=!rosult.fastEqu<i1s(p1an)H

queryExecutionMetries.incNumEffectiveExecution(rule.ruleName),

quer^^ExecutionMetrics.incTimeEffectiveExecutionB)y(rule.ruleNam

,runTime)

「)I⑴(h，ng(、l」)lg(T.I(以1八11。(「"k?門ih\，imc,pkn.rm：：

resul

iteration+=

if(iteration>ba二ch.strategy,maxIterations)

valendingMsg=if(batch,strategy.maxIterationsSetting==nul1)

}else

s”,pleaseset'${batch,strategy.maxIterationsSetting}'toalarger]

valmessage=s"Maxiterations(${iteration-1)reachedforbatch$

)atch.name)H

if(Utils.isTestingbatch,strategy.errorOnExceed)

IhrownewTreeNodeExceplion(curPlan,message,

1ogWarning(message)

if(batch,strategy二二Once

Uli】s.isTesling&&!blacklisledOnceBalches.contains(batch,name))

continue=fals

if(curPlan.fastEquals(lastPlan))

logTrace

s“Fixedpointreachedforbatch${batch,name}after${iteration?-1)i

continue=fals

lastPlan=curPlad

planChangeLogger.logBatch(batch,name,batchStartPlan,curPlan)

planChangeLogger.logMetrics(RuleExecutor.getCurrentMetrics()-befor

Metrics)

curPlan

如上代碼的主要處理過程如下：

1、遍歷的Analyzer類中的batches列表：

通過batches方法獲取所有的catalog綁定相關(guān)的規(guī)則，在Analyzer中包括

Substitution、Hints、Resolution、UDF、Subquery等幾個(gè)規(guī)則組；

以較為常見的"Resolution”規(guī)則組為例，其具有非常多的規(guī)則用于解析函數(shù)、

Namespace,數(shù)據(jù)表、視圖、列等信息，當(dāng)然用戶也可以子定義相關(guān)規(guī)貝U：

ResolvcTableValuedFunctions:

ResolveNamespace(catal.ogManager):：■

newResolveCatalogs(catalogManager)::

ResolveRelations:：

RescdveTables:：

ResolveReferences：：

ResolveCreateNamedStruct：

ResolvcDeserializer：

ResolveNewInstance:

ResolveUpCast:：

ResolveGroupingAnalytics：

ResolvePivot：：

ResolveOrdinannOrderByAndGroupBy:：，

ResolveAggAliasInGroupBy:

其中，Batch類的定義如下，包括Batch名稱、循環(huán)執(zhí)行策略、具體的規(guī)則組

集合，循環(huán)執(zhí)行策略Strategy又分為Once和FixedPoint兩種，即僅執(zhí)行一次

和固定次數(shù):

rotectedcaseclassBatch(name：String,strategy：Strategy,rules：Rule

2、將每個(gè)Batch中所有的規(guī)則Rule對(duì)象實(shí)施于該UnsolvedLogicalPlan,并

且該Batch中規(guī)則可能要執(zhí)行多輪，直到執(zhí)行的批數(shù)等于

batch,strategy,maxIterations或者logicalplan與上個(gè)批次的結(jié)果比沒有變

化，則退出執(zhí)行；

其中在Spark中的定義如下，在spark3.0中默認(rèn)可最大循環(huán)100次：

conf,analyzerYax11erations,

errorOnExceed二

maxIterationsSetting=SQLConf.ANALYZER_MAX_ITERATIONS.key

■

|.interna1()

|.：：,‘’―…Il」.III，，.「,」「i：」：：」，!.「「「「TII:”:■■■??■??■；

.creatcWithDefault(100)

接下來以將ResolveRelations（解析數(shù)據(jù)表或者視圖）規(guī)則應(yīng)用于Unresolved

LogicalPlan的解析過程為例，支持解析UnresolvedRelation>

UnresolvedTable^Unreso1vcdTab1eOrView等多種未解析的數(shù)據(jù)源：

|defapply(plan：LogicalPlan)：LogicalPlan:RBSolveTempViews(plan).reso|

lookupRelation(u.multipartldentifier).map(resolveViews).getOrElse(u

caseu@UnresolvedTable(identifier)二

u.fai1Analysis(s"${v.identifier.quoted}isaviewnottable.")

casetable=>table

}.getOrElse(u)

caseu@UnresolvedTab1eOrView(identifier)=

1ookupTab1eOrView(identifier).getOrElse(u)

當(dāng)解析對(duì)象為UnresolvedRelation實(shí)例時(shí)，調(diào)生lookupRelation方法來對(duì)其

進(jìn)行解析，通過SessionCatalog或者擴(kuò)展的CatalogPlugin來獲取數(shù)據(jù)源的元

數(shù)據(jù)，并生成ResolvedLogicalPlan：

rivatedeflookupRelation(identifier：Seq[String])：Option[LogicalPlan|

expandRelationName[identifier)match

caseSessionCatalogAndldentifier(catalog,ident)二

lazyvalloaded=CatalogV2Util.loadTable(catalog,ident).map

vlSessionCatalog.getRelation(vlTable.

casetable

SubqueryAlias

catalog,name+：ident.asMultipartldcntificr,^^^^^^^^^^^^!

DataSourceV2Relation.create(table,Some(catalog),Some(ident)))

最常見的是SessionCatalog,作為SparkSession級(jí)別catalog接口對(duì)象,其

定義如下，包括Externalcatalog、G1obalTempViewManager

FunctionRegistrySQLConf、HadoopConfiguration>Parser.

FunctionResourceLoader對(duì)象；其中，Externalcatalog有兩個(gè)主要的實(shí)現(xiàn)

類：HiveExternalCatalog和InMcmoryCatalog,而HiveExternalCatalog則主

要應(yīng)用于企業(yè)級(jí)的業(yè)務(wù)場(chǎng)景中:

LassSessionCatalog

globalTcmpViewManagerBuilder：()二)GlobalTcmpViowManager,

functionRegistry：

hadoopConf：Config」ration,

parser：ParserInterface,

>i：l|.''i1.Ir).：：,1:i,^?：-','.''：

如果采用默認(rèn)的SessionCatalog,當(dāng)需要獲取數(shù)據(jù)表時(shí)則通過

Externalcatalog實(shí)例調(diào)用其對(duì)應(yīng)的接口來實(shí)現(xiàn)：

werridedefloadTableGdent：Identifier)：Table={■

va]catalogTable二try

catalog.gctTableMctadata(idcnt.asTableldentifier)

、L,「「'?1」：常

D

VlTable(catalogTable)

■

lefgetTableMetadata(name：TableTdentifier)：CatalogTable

yaldb=formatDatabaseName(name,database.getOrElse(getCurrentDatabas

valtable二formatTableName(name.

requireDbExists(db)

requireTableExists(Tableldentifier(table,Some(db)))

接下來如果采用Externalcatalog接口的實(shí)現(xiàn)類HiveExternalCatalog的情況

下，則通過HiveClientlmpl類從Hive的metadata中類獲取用戶表的元數(shù)據(jù)相

關(guān)信息：

privatedefgetRawTableOption(dbName：String,tableName：String)：Option]

Option(client.getTable(dbName,tableName,false*don。Ilhnwoxcepli(

另外，如需擴(kuò)展的catalog范圍可通過實(shí)現(xiàn)CatalogPlugin接口、并且配置

uspark,sql.catalog.spark_catalogv參數(shù)來實(shí)現(xiàn)，例如在iceberg數(shù)據(jù)湖的

實(shí)現(xiàn)中通過自定義其catalog來實(shí)現(xiàn)其個(gè)性化的纓輯：

|spark.sql.catalog.spark_catalog

prg.apache,iceberg,spark.SparkSessionCatalog

3、返回解析后的ResolvedLogicalPlan?

以上處理邏輯中所涉及的主要的類之間的關(guān)系如下所示:

接下來仍然以前面的SQL語(yǔ)句（selectcollfromtabnamewhereco12>

10）為例，簡(jiǎn)要闡述如何將一個(gè)UnresolvedLogicalPlan解析成為Analyzed

LogicalPlan：

1、根據(jù)Analyzer的解析規(guī)則，UnResolvedRelalion節(jié)點(diǎn)可以應(yīng)用到

ResolveRelations規(guī)則，通過CatalogManger獲取數(shù)據(jù)源中表的信息，得到

Relation的相關(guān)列的信息并加上標(biāo)號(hào)，同時(shí)創(chuàng)建一個(gè)針對(duì)數(shù)據(jù)表的

SubqucryAlias節(jié)點(diǎn);

2、針對(duì)過濾條件col2>10的過濾條件，針對(duì)列UnresolvedAttribute可以適

用到ResolveReference規(guī)則,根據(jù)第1步中得到的列信息可以進(jìn)行解析；數(shù)字

10可以應(yīng)用到ImplicitTypcCasts規(guī)則對(duì)該數(shù)字匹配最合適的數(shù)據(jù)類型；

3、針對(duì)Project節(jié)點(diǎn)，接下來在進(jìn)行下一輪解析，再次匹配到

ResolveReference規(guī)則對(duì)投影列進(jìn)行解析，從而將整棵樹解析為Resolved

LogicalPlan0

生成OptimizedLogicalPlan

得到ResolvedLogicalPlan之后，為了使SQL語(yǔ)句的執(zhí)行性能更優(yōu)，則需要根

據(jù)一些規(guī)則進(jìn)一步優(yōu)化邏輯計(jì)劃樹,生成OptimizedLogicalPlan。

本文采用的是Spark3.0的源碼，生成OptimizedLogicalPlan是通過懶加載

的方式被調(diào)用的，并且Optimizer類與Analyzer類一樣繼承了RuleExecutor

類，所有基于規(guī)則(RB0)的優(yōu)化實(shí)際都是通過RuleExecutor類來執(zhí)行，同樣也

是將所有規(guī)則構(gòu)建為多個(gè)批次，并且將所有批次中規(guī)則應(yīng)用于Analyzed

LogicalPlan,直到樹不再改變或者執(zhí)行優(yōu)化的循環(huán)次數(shù)超過最大限制

(spark.sql.optimizer,maxIterations,默認(rèn)100):

11azyvaloptimizedPlan：LogicalPlan二executePhase(QueryPlanningTracke

optimizingandplanning.

|valplan=sparkSession.sessionState.optimizer.executcAndTrack(withCaj

thedData.clone。，tracker)

■

|defexecuteAndTrack(plan：TrueType,tracker：QueryPlanningTracker)：Tr

|QueryPlanningTracker.withTracker(tracker)

|execute(plan)

邏輯計(jì)劃優(yōu)化規(guī)則仍然又多個(gè)Batch組成，每個(gè)Balch中包含多個(gè)具體的Rule

并且可以執(zhí)行一次或者固定次數(shù)。其中比較常用的優(yōu)化規(guī)則有：謂詞下推、常

量累加、列剪枝等幾種。

謂詞下推將盡可能使得謂詞計(jì)算靠近數(shù)據(jù)源，根據(jù)不同的場(chǎng)景有

LimitPushDown、PushProjectionThroughUnionPushDownPredicates等多種

實(shí)現(xiàn)，PushDownPredicates又包含PushPredicatcThroughNonJoin和

PushPredicateThroughJoin；

其中，PushPredicateThroughJoin可實(shí)現(xiàn)將謂詞計(jì)算下推至join算子的下

面，從而可以提升數(shù)據(jù)表之間的join計(jì)算過程中所帶來的網(wǎng)絡(luò)、內(nèi)存以及10

等性能開銷：

alapplyLocally:PartialFunction[LogicalPlan,LogicalPlan]:

|casef@Filter(filterCondition,Join(lefttright,joinType,joinConditi|

|on,hint)

|val(leftFi1terConditions,rightFi1terConditions,commonFi1terConditi

split(splitConjur.ctivePredicates(fi1terCondition),left,right)Kn

joinTypematch

case_：TnnerLike二

valncwLeft=1eftFilterConditions.

rcduceLcftOption(And).