Spark SQL之External DataSource外部数据源

本文主要是介绍Spark SQL之External DataSource外部数据源，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

http://blog.csdn.net/oopsoom/article/details/42061077

一、Spark SQL External DataSource简介

随着Spark1.2的发布，Spark SQL开始正式支持外部数据源。Spark SQL开放了一系列接入外部数据源的接口，来让开发者可以实现。

这使得Spark SQL支持了更多的类型数据源，如json, parquet, avro, csv格式。只要我们愿意，我们可以开发出任意的外部数据源来连接到Spark SQL。之前大家说的支持HBASE，Cassandra都可以用外部数据源的方式来实现无缝集成。

二、External DataSource

拿Spark1.2的json为例，它支持已经改为了实现了外部数据源的接口方式。所以除了先前我们操作json的API，又多了一种DDL创建外部数据源的方式。

parquetFile的操作方式也如下类似，就不一一列举了。

2.1 SQL方式 CREATE TEMPORARY TABLE USING OPTIONS

在Spark1.2之后，支持了一种CREATE TEMPORARY TABLE USING OPTIONS的DDL语法来创建外部数据源的表。

[sql]  view plain copy 
CREATE TEMPORARY TABLE jsonTable  
USING org.apache.spark.sql.json  
OPTIONS (  
  path '/path/to/data.json'  
)  

1、操作示例：

我们拿example下people.json文件来做示例。

[java]  view plain copy 
shengli-mac$ cat /Users/shengli/git_repos/spark/examples/src/main/resources/people.json  
{"name":"Michael"}  
{"name":"Andy", "age":30}  
{"name":"Justin", "age":19}  

2、DDL创建外部数据源表jsonTable：

[java]  view plain copy 
14/12/21 16:32:14 INFO repl.SparkILoop: Created spark context..  
Spark context available as sc.  
  
scala> import org.apache.spark.sql.SQLContext  
import org.apache.spark.sql.SQLContext  
  
scala> val sqlContext  = new SQLContext(sc)  
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@7be62956  
  
scala> import sqlContext._  
import sqlContext._  
//创建jsonTable外部数据源表，并且指定其数数据源文件是people.json这个json文件，同时指定使用org.apache.spark.sql.json该类型的隐式转化类(这个后续文章会介绍)  
scala> val jsonDDL = s"""  
     | |CREATE TEMPORARY TABLE jsonTable  
     | |USING org.apache.spark.sql.json  
     | |OPTIONS (  
     | | path  'file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json'  
     | |)""".stripMargin  
jsonDDL: String =   
"  
CREATE TEMPORARY TABLE jsonTable  
USING org.apache.spark.sql.json  
OPTIONS (  
 path  'file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json'  
)"  
  
scala> sqlContext.sql(jsonDDL).collect() //创建该外部数据源表jsonTable  
14/12/21 16:44:27 INFO scheduler.DAGScheduler: Job 0 finished: reduce at JsonRDD.scala:57, took 0.204461 s  
res0: Array[org.apache.spark.sql.Row] = Array()  

我们来看下该schemaRDD:

[java]  view plain copy 
scala> val jsonSchema = sqlContext.sql(jsonDDL)  
jsonSchema: org.apache.spark.sql.SchemaRDD =   
SchemaRDD[7] at RDD at SchemaRDD.scala:108  
== Query Plan ==  
== Physical Plan ==  
ExecutedCommand (CreateTableUsing jsonTable, org.apache.spark.sql.json, Map(path -> file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json))  

ExecutedCommand来取把数据用spark.sql.json的方式从path加载到jsonTable中。涉及到得类是CreateTableUsing，后续源码分析会讲到。

各阶段执行计划情况：

[java]  view plain copy 
scala> sqlContext.sql("select * from jsonTable").queryExecution  
res6: org.apache.spark.sql.SQLContext#QueryExecution =   
== Parsed Logical Plan ==  
'Project [*]  
 'UnresolvedRelation None, jsonTable, None  
  
== Analyzed Logical Plan ==  
Project [age#0,name#1]  
 Relation[age#0,name#1] JSONRelation(file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json,1.0)  
  
== Optimized Logical Plan ==  
Relation[age#0,name#1] JSONRelation(file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json,1.0)  
  
== Physical Plan ==  
PhysicalRDD [age#0,name#1], MapPartitionsRDD[27] at map at JsonRDD.scala:47  
  
Code Generation: false  
== RDD ==  

至此，创建加载外部数据源到Spark SQL已经完成。

我们可以使用任何我们希望的方式来查询：

3、SQL查询方式：

[java]  view plain copy 
scala> sqlContext.sql("select * from jsonTable")  
21 16:52:13 INFO spark.SparkContext: Created broadcast 6 from textFile at JSONRelation.scala:39  
res2: org.apache.spark.sql.SchemaRDD =   
SchemaRDD[20] at RDD at SchemaRDD.scala:108  
== Query Plan ==  
== Physical Plan ==  
PhysicalRDD [age#2,name#3], MapPartitionsRDD[24] at map at JsonRDD.scala:47  

执行查询：

scala> sqlContext.sql("select * from jsonTable").collect()
res1: Array[org.apache.spark.sql.Row] = Array([null,Michael], [30,Andy], [19,Justin])

2.2 API方式

sqlContext.jsonFile

[java]  view plain copy 
scala> val json = sqlContext.jsonFile("file:///Users/shengli/git_repos/spark/examples/src/main/resources/people.json")  
scala> json.registerTempTable("jsonFile")  
  
scala> sql("select * from jsonFile").collect()  
res2: Array[org.apache.spark.sql.Row] = Array([null,Michael], [30,Andy], [19,Justin])  

总的来说，Spark SQL 在努力的向各种数据源靠拢，希望让Spark SQL能和其它许多类型的数据源的集成。

Spark SQL提供的了一种创建加载外部数据源表的DDL语法：CREATE TEMPORARY TABLE USING OPTIONS

Spark SQL对外开放了一系列的扩展接口，能够通过实现这些接口，来实现对不同的数据源接入，如avro, csv, parquet,json, etc

三、Sources包核心

Spark SQL在Spark1.2中提供了External DataSource API，开发者可以根据接口来实现自己的外部数据源，如avro, csv, json, parquet等等。

在Spark SQL源代码的org/spark/sql/sources目录下，我们会看到关于External DataSource的相关代码。这里特别介绍几个:

1、DDLParser

专门负责解析外部数据源SQL的SqlParser，解析create temporary table xxx using options (key 'value', key 'value') 创建加载外部数据源表的语句。

[java]  view plain copy 
protected lazy val createTable: Parser[LogicalPlan] =  
   CREATE ~ TEMPORARY ~ TABLE ~> ident ~ (USING ~> className) ~ (OPTIONS ~> options) ^^ {  
     case tableName ~ provider ~ opts =>  
       CreateTableUsing(tableName, provider, opts)  
   }  

2、CreateTableUsing

一个RunnableCommand，通过反射从外部数据源lib中实例化Relation，然后注册到为temp table。

[java]  view plain copy 
private[sql] case class CreateTableUsing(  
    tableName: String,  
    provider: String,  // org.apache.spark.sql.json   
    options: Map[String, String]) extends RunnableCommand {  
  
  def run(sqlContext: SQLContext) = {  
    val loader = Utils.getContextOrSparkClassLoader  
    val clazz: Class[_] = try loader.loadClass(provider) catch { //do reflection  
      case cnf: java.lang.ClassNotFoundException =>  
        try loader.loadClass(provider + ".DefaultSource") catch {  
          case cnf: java.lang.ClassNotFoundException =>  
            sys.error(s"Failed to load class for data source: $provider")  
        }  
    }  
    val dataSource = clazz.newInstance().asInstanceOf[org.apache.spark.sql.sources.RelationProvider] //json包DefaultDataSource  
    val relation = dataSource.createRelation(sqlContext, new CaseInsensitiveMap(options))//创建JsonRelation  
  
    sqlContext.baseRelationToSchemaRDD(relation).registerTempTable(tableName)//注册  
    Seq.empty  
  }  
}  

2、DataSourcesStrategy

在 Strategy 一文中，我已讲过Streategy的作用，用来Plan生成物理计划的。这里提供了一种专门为了解析外部数据源的策略。

最后会根据不同的BaseRelation生产不同的PhysicalRDD。不同的BaseRelation的scan策略下文会介绍。

[java]  view plain copy 
private[sql] object DataSourceStrategy extends Strategy {  
  def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {  
    case PhysicalOperation(projectList, filters, l @ LogicalRelation(t: CatalystScan)) =>  
      pruneFilterProjectRaw(  
        l,  
        projectList,  
        filters,  
        (a, f) => t.buildScan(a, f)) :: Nil  
    ......  
    case l @ LogicalRelation(t: TableScan) =>  
      execution.PhysicalRDD(l.output, t.buildScan()) :: Nil  
  
    case _ => Nil  
  }  

3、interfaces.scala

该文件定义了一系列可扩展的外部数据源接口，对于想要接入的外部数据源，我们只需实现该接口即可。里面比较重要的trait RelationProvider 和 BaseRelation，下文会详细介绍。

4、filters.scala

该Filter定义了如何在加载外部数据源的时候，就进行过滤。注意哦，是加载外部数据源到Table里的时候，而不是Spark里进行filter。这个有点像hbase的coprocessor，查询过滤在Server上就做了，不在Client端做过滤。

5、LogicalRelation

封装了baseRelation，继承了catalyst的LeafNode,实现MultiInstanceRelation。

四、External DataSource注册流程

用spark sql下sql/json来做示例, 画了一张流程图，如下：

注册外部数据源的表的流程：

1、提供一个外部数据源文件，比如json文件。

2、提供一个实现了外部数据源所需要的interfaces的类库，比如sql下得json包，在1.2版本后改为了External Datasource实现。

3、引入SQLContext，使用DDL创建表，如create temporary table xxx using options (key 'value', key 'value')

4、External Datasource的DDLParser将对该SQL进行Parse

5、Parse后封装成为一个CreateTableUsing类的对象。该类是一个RunnableCommand，其run方法会直接执行创建表语句。

6、该类会通过反射来创建一个org.apache.spark.sql.sources.RelationProvider，该trait定义要createRelation，如json，则创建JSONRelation，若avro,则创建AvroRelation。

7、得到external releation后，直接调用SQLContext的baseRelationToSchemaRDD转换为SchemaRDD

8、最后registerTempTable(tableName) 来注册为Table，可以用SQL来查询了。

五、External DataSource解析流程

先看图，图如下：

Spark SQL解析SQL流程如下：

1、Analyzer通过Rule解析，将UnresolvedRelation解析为JsonRelation。

2、通过Parse，Analyzer，Optimizer最后得到JSONRelation(file:///path/to/shengli.json,1.0)

3、通过sources下得DataSourceStrategy将LogicalPlan映射到物理计划PhysicalRDD。

4、PhysicalRDD里包含了如何查询外部数据的规则，可以调用execute()方法来执行Spark查询。

六、External Datasource Interfaces

在第一节我已经介绍过，主要的interfaces，主要看一下BaseRelation和RelationProvider。

如果我们要实现一个外部数据源，比如avro数据源，支持spark sql操作avro file。那么久必须定义AvroRelation来继承BaseRelation。同时也要实现一个RelationProvider。

BaseRelation：

是外部数据源的抽象，里面存放了 schema的映射，和如何scan数据的规则。

[java]  view plain copy 
abstract class BaseRelation {  
  def sqlContext: SQLContext  
  def schema: StructType  

[java]  view plain copy 
abstract class PrunedFilteredScan extends BaseRelation {  
  def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row]  
}  

1、schema我们如果自定义Relation，必须重写schema，就是我们必须描述对于外部数据源的Schema。

2、buildScan我们定义如何查询外部数据源，提供了4种Scan的策略，对应4种BaseRelation。

我们支持4种BaseRelation，分为TableScan， PrunedScan，PrunedFilterScan，CatalystScan。

1、 TableScan：

默认的Scan策略。

2、 PrunedScan：

这里可以传入指定的列，requiredColumns，列裁剪，不需要的列不会从外部数据源加载。

3、 PrunedFilterScan：

在列裁剪的基础上，并且加入Filter机制，在加载数据也的时候就进行过滤，而不是在客户端请求返回时做Filter。

4、 CatalystScan：

Catalyst的支持传入expressions来进行Scan。支持列裁剪和Filter。

RelationProvider：

我们要实现这个，接受Parse后传入的参数，来生成对应的External Relation，就是一个反射生产外部数据源Relation的接口。

[java]  view plain copy 
trait RelationProvider {  
  /** 
   * Returns a new base relation with the given parameters. 
   * Note: the parameters' keywords are case insensitive and this insensitivity is enforced 
   * by the Map that is passed to the function. 
   */  
  def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation  
}  

七、External Datasource定义示例

在Spark1.2之后，json和parquet也改为通过实现External API来进行外部数据源查询的。

下面以json的外部数据源定义为示例，说明是如何实现的：

1、JsonRelation

定义处理对于json文件的，schema和Scan策略，均基于JsonRDD，细节可以自行阅读JsonRDD。

[java]  view plain copy 
private[sql] case class JSONRelation(fileName: String, samplingRatio: Double)(  
    @transient val sqlContext: SQLContext)  
  extends TableScan {  
  
  private def baseRDD = sqlContext.sparkContext.textFile(fileName) //读取json file  
  
  override val schema =  
    JsonRDD.inferSchema(  // jsonRDD的inferSchema方法，能自动识别json的schema，和类型type。  
      baseRDD,  
      samplingRatio,  
      sqlContext.columnNameOfCorruptRecord)  
  
  override def buildScan() =  
    JsonRDD.jsonStringToRow(baseRDD, schema, sqlContext.columnNameOfCorruptRecord) //这里还是JsonRDD，调用jsonStringToRow查询返回Row  
}  

2、DefaultSource

parameters中可以获取到options中传入的path等自定义参数。

这里接受传入的参数，来狗仔JsonRelation。

[java]  view plain copy 
private[sql] class DefaultSource extends RelationProvider {  
  /** Returns a new base relation with the given parameters. */  
  override def createRelation(  
      sqlContext: SQLContext,  
      parameters: Map[String, String]): BaseRelation = {  
    val fileName = parameters.getOrElse("path", sys.error("Option 'path' not specified"))  
    val samplingRatio = parameters.get("samplingRatio").map(_.toDouble).getOrElse(1.0)  
  
    JSONRelation(fileName, samplingRatio)(sqlContext)  
  }  
}  