sparkSQL1.1入门之四：深入了解sparkSQL执行计划

本章就着重介绍一个工具hive/console，来加深读者对sparkSQL的执行计划的理解。

该工具是给开发人员使用，在编译生成的安装部署包中并没有；该工具须要使用sbt编译执行。要使用该工具，须要具备下面条件：

• spark1.1.0源代码
• hive0.12源代码并编译
• 配置环境变量

ant clean package -Dhadoop.version=2.2.0 -Dhadoop-0.23.version=2.2.0 -Dhadoop.mr.rev=23

export HIVE_HOME=/app/hadoop/hive012/src/build/dist
export HIVE_DEV_HOME=/app/hadoop/hive012/src
export HADOOP_HOME=/app/hadoop/hadoop220

sbt/sbt hive/console

 /*源自 sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala */
  // The test tables that are defined in the Hive QTestUtil.
  // /itests/util/src/main/java/org/apache/hadoop/hive/ql/QTestUtil.java
  val hiveQTestUtilTables = Seq(
    TestTable("src",
      "CREATE TABLE src (key INT, value STRING)".cmd,
      s"LOAD DATA LOCAL INPATH ‘${getHiveFile("data/files/kv1.txt")}‘ INTO TABLE src".cmd),
    TestTable("src1",
      "CREATE TABLE src1 (key INT, value STRING)".cmd,
      s"LOAD DATA LOCAL INPATH ‘${getHiveFile("data/files/kv3.txt")}‘ INTO TABLE src1".cmd),
    TestTable("srcpart", () => {
      runSqlHive(
        "CREATE TABLE srcpart (key INT, value STRING) PARTITIONED BY (ds STRING, hr STRING)")
      for (ds <- Seq("2008-04-08", "2008-04-09"); hr <- Seq("11", "12")) {
        runSqlHive(
          s"""LOAD DATA LOCAL INPATH ‘${getHiveFile("data/files/kv1.txt")}‘
             |OVERWRITE INTO TABLE srcpart PARTITION (ds=‘$ds‘,hr=‘$hr‘)
           """.stripMargin)
      }
    }),
......
)

 /*源自 sql/hive/src/main/scala/org/apache/spark/sql/hive/TestHive.scala */
  /** The location of the compiled hive distribution */
  lazy val hiveHome = envVarToFile("HIVE_HOME")
  /** The location of the hive source code. */
  lazy val hiveDevHome = envVarToFile("HIVE_DEV_HOME")

 /* 源自 project/SparkBuild.scala */
object Hive {
  lazy val settings = Seq(
    javaOptions += "-XX:MaxPermSize=1g",
    // Multiple queries rely on the TestHive singleton. See comments there for more details.
    parallelExecution in Test := false,
    // Supporting all SerDes requires us to depend on deprecated APIs, so we turn off the warnings
    // only for this subproject.
    scalacOptions <<= scalacOptions map { currentOpts: Seq[String] =>
      currentOpts.filterNot(_ == "-deprecation")
    },
    initialCommands in console :=
      """
        |import org.apache.spark.sql.catalyst.analysis._
        |import org.apache.spark.sql.catalyst.dsl._
        |import org.apache.spark.sql.catalyst.errors._
        |import org.apache.spark.sql.catalyst.expressions._
        |import org.apache.spark.sql.catalyst.plans.logical._
        |import org.apache.spark.sql.catalyst.rules._
        |import org.apache.spark.sql.catalyst.types._
        |import org.apache.spark.sql.catalyst.util._
        |import org.apache.spark.sql.execution
        |import org.apache.spark.sql.hive._
        |import org.apache.spark.sql.hive.test.TestHive._
        |import org.apache.spark.sql.parquet.ParquetTestData""".stripMargin
  )
}

//在控制台逐行执行
case class Person(name:String, age:Int, state:String)
sparkContext.parallelize(Person("Michael",29,"CA")::Person("Andy",30,"NY")::Person("Justin",19,"CA")::Person("Justin",25,"CA")::Nil).registerTempTable("people")
val query= sql("select * from people")

query.printSchema

query.queryExecution

query.queryExecution.logical

query.queryExecution.analyzed

query.queryExecution.optimizedPlan

query.queryExecution.sparkPlan

query.toDebugString

以下看看这些数据源的schema：

{  
   "fullname": "Sean Kelly",     
   "org": "SK Consulting",     
   "emailaddrs": [     
      {"type": "work", "value": "kelly@seankelly.biz"},     
      {"type": "home", "pref": 1, "value": "kelly@seankelly.tv"}     
   ],     
    "telephones": [     
      {"type": "work", "pref": 1, "value": "+1 214 555 1212"},     
      {"type": "fax", "value": "+1 214 555 1213"},     
      {"type": "mobile", "value": "+1 214 555 1214"}     
   ],     
   "addresses": [     
      {"type": "work", "format": "us",     
       "value": "1234 Main StnSpringfield, TX 78080-1216"},     
      {"type": "home", "format": "us",     
       "value": "5678 Main StnSpringfield, TX 78080-1316"}     
   ],     
    "urls": [     
      {"type": "work", "value": "http://seankelly.biz/"},     
      {"type": "home", "value": "http://seankelly.tv/"}     
   ]     
}

jsonFile("/home/mmicky/data/nestjson.json").registerTempTable("jsonPerson")
val jsonQuery = sql("select * from jsonPerson")

jsonQuery.printSchema

jsonQuery.queryExecution

parquetFile("/home/mmicky/data/spark/wiki_parquet").registerTempTable("parquetWiki")
val parquetQuery = sql("select * from parquetWiki")

parquetQuery.printSchema

parquetQuery.queryExecution

以下我们使用sales表看看其schema和整个执行计划。首先定义一个查询hiveQuery：

val hiveQuery = sql("select * from sales")

hiveQuery.printSchema

hiveQuery.queryExecution

sql("select state,avg(age) from people group by state").queryExecution

sql("select state,avg(age) from people group by state").toDebugString

sql("select a.name,b.name from people a join people b where a.name=b.name").queryExecution

sql("select a.name,b.name from people a join people b where a.name=b.name").toDebugString

sql("select distinct a.name,b.name from people a join people b where a.name=b.name").queryExecution

sql("select distinct a.name,b.name from people a join people b where a.name=b.name").toDebugString

sql("select name from (select * from people where age >=19) a where a.age <30").queryExecution

sql("select name from (select name,state as location from people) a where location=‘CA‘").queryExecution

sql("select name,1+2 from people").queryExecution

object CombineFilters extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    case Filter(c1, Filter(c2, grandChild)) =>
      Filter(And(c1,c2),grandChild)
  }
}

val query= sql("select * from people").where(‘age >=19).where(‘age <30)
query.queryExecution.analyzed

CombineFilters(query.queryExecution.analyzed)

val hiveQuery = sql("SELECT * FROM (SELECT * FROM src) a")
hiveQuery.queryExecution.analyzed

hiveQuery.queryExecution.analyzed transform {
   case Project(projectList, child) if projectList == child.output => child
 }