主要原理

spark sql 核心：
ParseInterface：
專門負責解析外部資料來源SQL的SqlParser。目前自帶的parser已經能滿足各種需求
RunnableCommand：
從反射的資料來源中例項化relation，然後註冊到temp table中。
Strategy：
將plan對映為物理計劃。
RelationProvider：
提供一個Relation。
BaseRelation：
可提供sql的一些邏輯操作（insert等）。

Spark SQL解析SQL流程如下：
1、Analyzer通過Rule解析，將UnresolvedRelation解析為JsonRelation。
2、通過Parse，Analyzer，Optimizer最後得到Relation（_*）
3、自定義的Strategy將LogicalPlan對映到物理計劃PhysicalRDD。
4、PhysicalRDD裡包含了如何查詢外部資料的規則。

簡單示例

使用自定義的relation，strategy，plan實現簡單的select，insert語句

程式碼：

Relation


case class TextRelation(sqlContext: SQLContext, schema: StructType, path: String) extends BaseRelation with InsertableRelation {
override def insert(data: DataFrame, overwrite: Boolean): Unit = {
if (!new File(path).exists())
data.rdd.map(_.mkString(",")).saveAsTextFile(path)
}
}


Source


class TextSource extends SchemaRelationProvider {
override def createRelation(sqlContext: SQLContext, parameters: Map[String, String], schema: StructType): BaseRelation = {
val path = parameters.getOrElse("path", "/home/wpy/tmp/external_sql/testSql")
TextRelation(sqlContext, schema, path)
}
}


sql入口


class Text4SQLContext(sc: SparkContext, sqlContext: SQLContext){
sqlContext.experimental.extraStrategies = new TextStrategies().TextStrategy :: Nil
def sql(sqlText: String): DataFrame = {
sqlContext.sql(sqlText)
}
}


Plan


case class LogicalText(output: Seq[Attribute], path: String) extends LogicalPlan {
override def children: Seq[LogicalPlan] = Nil
}


case class PhysicalText(output: Seq[Attribute], path: String) extends SparkPlan {
override protected def doExecute(): RDD[InternalRow] = {
sparkContext.textFile(path).map { row =>
val fields = row.split(",").map(UTF8String.fromString)
UnsafeProjection.create(schema)(InternalRow.fromSeq(fields))
}
}
override def children: Seq[SparkPlan] = Nil
}


case class TextExecuteCommand(cmd: RunnableCommand) extends SparkPlan {
override protected def doExecute(): RDD[InternalRow] = {
ExecutedCommandExec(cmd).execute()
}
override def output: Seq[Attribute] = cmd.output
override def children: Seq[SparkPlan] = Nil
}


Strategy


class TextStrategies extends QueryPlanner[SparkPlan] with PredicateHelper {
override def strategies: Seq[GenericStrategy[SparkPlan]] = TextStrategy :: Nil
object TextStrategy extends Strategy {
override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
plan match {
case LogicalText(output, path) => PhysicalText(output, path) :: Nil
case LogicalRelation(TextRelation(_, _, path), output, _) => PhysicalText(output, path) :: Nil
case [email protected]([email protected](t: TextRelation, _, _), part, query, overwrite, false) if part.isEmpty =>
ExecutedCommandExec(InsertIntoDataSourceCommand(l, query, overwrite)) :: Nil
case _ => Nil
}
}
}
override protected def collectPlaceholders(plan: SparkPlan): Seq[(SparkPlan, LogicalPlan)] = {
plan.collect {
case [email protected](logicalPlan) => placeholder -> logicalPlan
}
// Nil
}
override protected def prunePlans(plans: Iterator[SparkPlan]): Iterator[SparkPlan] = {
plans
}
}


示例：


object TestTextSql {
val conf = new SparkConf().setMaster("local[*]").setAppName(getClass.getCanonicalName)
val ss = SparkSession.builder().config(conf).getOrCreate()
def main(args: Array[String]): Unit = {
val sqlContext = ss.sqlContext
val sparkContext = ss.sparkContext
val ts = new Text4SQLContext(sparkContext, sqlContext)
ts.sql(
"""create table test1(
|word string,
|num string
|) using external.datasource.TextSource
|options(
|path '/home/wpy/tmp/external_sql/test1'
|)
""".stripMargin)
ts.sql("select * from test1").show
print("=============================================\n")
ts.sql(
"""create table test2(
|word string,
|num string
|) using external.datasource.TextSource
|options(
|path '/home/wpy/tmp/external_sql/test2'
|)
""".stripMargin)
ts.sql(
"""
|insert into table test2
|select * from test1
""".stripMargin)
ts.sql("select * from test2 order by word").show
}
}


資料來源
test1表（文字檔案）
1，a
2，b
3，c

測試結果如下：

+—-+—+
|word|num|
+—-+—+
| 1| a|
| 2| b|
| 3| c|
+—-+—+

=============================================

+—-+—+
|word|num|
+—-+—+
| 1| a|
| 2| b|
| 3| c|
+—-+—+

由於這部分程式碼較為簡單，此處則不再贅述，理論上自定義資料來源可以實現在任何能讀取的資料上進行sql操作，其中還可以自定義filter方法從源頭對資料進行過濾，從而實現對大量資料的快速查詢，注意此處使用了spark自帶的RDD，使用其他spark不支援的資料來源時，需要自行定義合適的RDD（物理計劃）去獲取資料。程式碼能直接執行（spark版本 2.2.0-SNAPSHOT）。