一绘面、開發(fā)一個spark應(yīng)用
//初始化sparksession
val spark = SparkSession.builder.appName("SparkSQL Test").master("local[4]").getOrCreate()
//transform操作,生成dataframe,可繼續(xù)執(zhí)行dataframe相關(guān)dsl api,
val sqlDf = spark.sql("select count(*) from table")
//action操作,spark-core開始執(zhí)行
sqlDf.show(false)
二、初始化sparksession-sessionState構(gòu)造過程
//1:sparksession懶加載sessionstate
lazy val sessionState: SessionState = {
parentSessionState
.map(_.clone(this))
.getOrElse {
val state = SparkSession.instantiateSessionState(
SparkSession.sessionStateClassName(sparkContext.conf),
self)
initialSessionOptions.foreach { case (k, v) => state.conf.setConfString(k, v) }
state
}
}
//2:實例化sessionstate
/**
* Helper method to create an instance of `SessionState` based on `className` from conf.
* The result is either `SessionState` or a Hive based `SessionState`.
*/
private def instantiateSessionState(
className: String,
sparkSession: SparkSession): SessionState = {
try {
// invoke `new [Hive]SessionStateBuilder(SparkSession, Option[SessionState])`
val clazz = Utils.classForName(className)
val ctor = clazz.getConstructors.head
//默認:BaseSessionStateBuilder, hive:HiveSessionStateBuilder
ctor.newInstance(sparkSession, None).asInstanceOf[BaseSessionStateBuilder].build()
...
//3: 構(gòu)建SessionState,初始化catalog侈沪、sqlparser揭璃、analyzer、optimzizer,內(nèi)置函數(shù)以及udf函數(shù)等等
def build(): SessionState = {
new SessionState(
session.sharedState,
conf,
experimentalMethods,
functionRegistry,
udfRegistration,
() => catalog,
sqlParser,
() => analyzer,
() => optimizer,
planner,
streamingQueryManager,
listenerManager,
() => resourceLoader,
createQueryExecution,
createClone)
}
注:QueryExecution亭罪、SessionState瘦馍、BaseSessionStateBuilder之間的關(guān)系:
(1)QueryExecution的analyzed、optimizedPlan是懶加載的,被調(diào)用時實際調(diào)用的是SessionState中的analyzer、optimizer的相關(guān)方法做解析和優(yōu)化
(2)SessionState的catalog、analyzer烘嘱、optimizer陋率、resourceLoader也是懶加載的,被調(diào)用時實際調(diào)用的是在BaseSessionStateBuilder初始化SessionState的時候生成的匿名函數(shù)
三试伙、transform-生成dataframe-resolved logicalPlan
/*
1: paserplan生成unresolved logicalPlan, ofRows方法中調(diào)用QueryExecution.assertAnalyzed(),
其實是sparkSession.sessionState.analyzer.executeAndCheck(logical),
再使用定義的各種解析規(guī)則,resolving unresolved attributes and relations,生成resolved logicalPlan,
最終new Dataset[Row](sparkSession, qe, RowEncoder(qe.analyzed.schema))生成dataframe*/
def sql(sqlText: String): DataFrame = {Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText))}
//2: 使用訪問者模式,astBuilder遍歷antlr sql語法樹,解析成catalyst的ast語法樹,生成unresolved的邏輯計劃
override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
astBuilder.visitSingleStatement(parser.singleStatement()) match {
case plan: LogicalPlan => plan
case _ =>
val position = Origin(None, None)
throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position)
}
}
//代碼3:將sql命令傳給antlr,使用SqlBase.g4生成的詞匯解析器SqlBaseLexer和語法解析器SqlBaseParser,對詞和語法校驗
protected def parse[T](command: String)(toResult: SqlBaseParser => T): T = {
logDebug(s"Parsing command: $command")
val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
lexer.removeErrorListeners()
lexer.addErrorListener(ParseErrorListener)
lexer.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced
val tokenStream = new CommonTokenStream(lexer)
val parser = new SqlBaseParser(tokenStream)
parser.addParseListener(PostProcessor)
parser.removeErrorListeners()
parser.addErrorListener(ParseErrorListener)
parser.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced
try {
try {
// first, try parsing with potentially faster SLL mode
parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
toResult(parser)
...
四曙咽、action-觸發(fā)執(zhí)行-優(yōu)化邏輯計劃,生成物理計劃,轉(zhuǎn)為rdd提交給sparkContex
//1:拉取20行數(shù)據(jù)到driver端,調(diào)用take(),最終調(diào)用head()
def show(): Unit = show(20)
def head(n: Int): Array[T] = withAction("head", limit(n).queryExecution)(collectFromPlan)
//2:Wrap一個action,監(jiān)控查詢執(zhí)行過程和時間花費,執(zhí)行用戶注冊的回調(diào)函數(shù)
private def withAction[U](name: String, qe: QueryExecution)(action: SparkPlan => U) = {
try {
/*觸發(fā)optimizer優(yōu)化器采用一系列優(yōu)化規(guī)則(eg:謂詞下推)對resolved logicalPlan進行優(yōu)化,
/sparkplanner選擇出最優(yōu)策略(eg:廣播表)將optimizedPlan轉(zhuǎn)化為sparkplan,
sparkplan應(yīng)用一系列規(guī)則,轉(zhuǎn)化為可預(yù)備執(zhí)行的物理計劃
*/
qe.executedPlan.foreach { plan =>
plan.resetMetrics()
}
val start = System.nanoTime()
val result = SQLExecution.withNewExecutionId(sparkSession, qe) {
//調(diào)用collectFromPlan,交給spark-core,執(zhí)行物理計劃,轉(zhuǎn)為rdd操作
action(qe.executedPlan)
}
val end = System.nanoTime()
sparkSession.listenerManager.onSuccess(name, qe, end - start)
result
} catch {
case e: Exception =>
sparkSession.listenerManager.onFailure(name, qe, e)
throw e
}
}
//3:QueryExecution中從優(yōu)化到生成可預(yù)備執(zhí)行的物理計劃工作流
lazy val optimizedPlan: LogicalPlan = sparkSession.sessionState.optimizer.execute(withCachedData)
lazy val sparkPlan: SparkPlan = {
SparkSession.setActiveSession(sparkSession)
// TODO: We use next(), i.e. take the first plan returned by the planner, here for now,
// but we will implement to choose the best plan.
planner.plan(ReturnAnswer(optimizedPlan)).next()
}
// executedPlan should not be used to initialize any SparkPlan. It should be
// only used for execution.
lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)
protected def prepareForExecution(plan: SparkPlan): SparkPlan = {
preparations.foldLeft(plan) { case (sp, rule) => rule.apply(sp) }
}
/** A sequence of rules that will be applied in order to the physical plan before execution. */
protected def preparations: Seq[Rule[SparkPlan]] = Seq(
python.ExtractPythonUDFs,
PlanSubqueries(sparkSession),
EnsureRequirements(sparkSession.sessionState.conf),
CollapseCodegenStages(sparkSession.sessionState.conf),
ReuseExchange(sparkSession.sessionState.conf),
ReuseSubquery(sparkSession.sessionState.conf))
//4:執(zhí)行自定義的回調(diào)函數(shù)函數(shù),該函數(shù)底層最終執(zhí)行sparkplan的do把物理計劃轉(zhuǎn)化為rdd操作
/**
* Collect all elements from a spark plan.
*/
private def collectFromPlan(plan: SparkPlan): Array[T] = {
// This projection writes output to a `InternalRow`, which means applying this projection is not
// thread-safe. Here we create the projection inside this method to make `Dataset` thread-safe.
val objProj = GenerateSafeProjection.generate(deserializer :: Nil)
plan.executeCollect().map { row =>
// The row returned by SafeProjection is `SpecificInternalRow`, which ignore the data type
// parameter of its `get` method, so it's safe to use null here.
objProj(row).get(0, null).asInstanceOf[T]
}
}
//5:將sparkplan轉(zhuǎn)為rdd,交給sparkContext提交job
/**
* Runs this query returning the result as an array.
*/
def executeCollect(): Array[InternalRow] = {
//getByteArrayRdd調(diào)用execute(),再調(diào)用doExecute()方法,將sparkplan轉(zhuǎn)為RDD
val byteArrayRdd = getByteArrayRdd()
val results = ArrayBuffer[InternalRow]()
//byteArrayRdd.collect()是rdd的action算子,會運行sc.runJob()提交job給spark集群
byteArrayRdd.collect().foreach { countAndBytes =>
decodeUnsafeRows(countAndBytes._2).foreach(results.+=)
}
results.toArray
}
