Spark SQL 代码简要阅读(基于Spark 1.1.0)

jopen 10年前

    Spark SQL允许相关的查询如SQL,HiveQL或Scala运行在spark上。其核心组件是一个新的RDD:SchemaRDD,SchemaRDDs由 行对象组成,并包含一个描述此行对象的每一列的数据类型的schema。SchemaRDD和传统关系型数据库的表类似。SchemaRDD可以通过已有 的RDD、Parquet(列式存储格式)类型文件、JSON数据集,或通过运行HiveQL获取存储在Apache Hive中的数据。社区文档介绍:https://spark.apache.org/docs/latest/sql-programming-guide.html

一个简单的例子

下面是一个使用Spark SQL的简单例子。

// sc is an existing SparkContext.

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD.

import sqlContext.createSchemaRDD

// Define the schema using a case class.

// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,

// you can use custom classes that implement the Product interface.

case class Person(name: String, age: Int)

// Create an RDD of Person objects and register it as a table.

val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt))

people.registerTempTable("people")

// SQL statements can be run by using the sql methods provided by sqlContext.

val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// The results of SQL queries are SchemaRDDs and support all the normal RDD operations.

// The columns of a row in the result can be accessed by ordinal.

teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

上述例子中定义了一个People类,通过加载people.txt文件并转换为一个SchemaRDD,然后调用registerTempTable方法注册为一个Table。后续便可以通过SQL对该表查询并生成新的SchemaRDDSchemaRDD继承于RDD,于是可以进行转换和Actions操作。

 

Spark SQL代码流程

从Spark SQL到RDD的DAG关系主要可以分为以下五步。

  1. 初始化,包括sqlContext,sqlContext包括Spark SQL执行的上下文与流程;定义并注册Table,定义Table的字段与类型,然后注册,注册实际上就是把Table的元数据存储在内存SimpleCatalog对象中。
  2. 解析SQL,并生成LogicalPlan(逻辑计划)。代码调用流程为:

    SQLContext.sql

    SQLContext .parseSql

    catalyst.SqlParser

    SqlLexical. Scanner

    最终通过SqlLexical. Scanner完成词法语法的解析并生成LogicalPlan。

     

  3. 由逻辑计划LogicalPlan生成QueryExecution。代码调用流程为:

    New SchemaRDDLike

    sqlContext.executePlan(baseLogicalPlan)

    生成QueryExecution

     

  4. QueryExecution转换为物理计划SparkPlan,代码调用流程为:

    SparkContext.runJob

    RDD.getDependencies

    SQLContext .QueryExecution.toRDD

    QueryExecution.prepareForExecution

    RuleExecutor. Apply

    Exchange.AddExchange. apply => SparkPlan

     

  5. 物理计划SparkPlan转换为RDD,通过调用SparkPlan.execute把树形结果的物理计划转换为RDD的DAG关系。

     

Spark SQL关键类图

其中右侧的LogicalPlan为逻辑计划,左边的SparkPlan为物理计划相关的类。

Spark SQL 代码简要阅读(基于Spark 1.1.0)

来自:http://www.cnblogs.com/shenh062326/p/4133501.html