Spark SQL之External DataSource外部数据源（二）源代码分析

    上周Spark1.2刚公布，周末在家没事，把这个特性给了解一下，顺便分析下源代码，看一看这个特性是怎样设计及实现的。

    /** Spark SQL源代码分析系列文章*/

  （Ps: External DataSource使用篇地址：Spark SQL之External DataSource外部数据源（一）演示样例 http://blog.csdn.net/oopsoom/article/details/42061077）

一、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') 创建载入外部数据源表的语句。

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。

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.getContextOrSparkClassLoaderval clazz: Class[_] = try loader.loadClass(provider) catch { //do reflectioncase 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包DefaultDataSourceval relation = dataSource.createRelation(sqlContext, new CaseInsensitiveMap(options))//创建JsonRelationsqlContext.baseRelationToSchemaRDD(relation).registerTempTable(tableName)//注冊Seq.empty} }

    2、DataSourcesStrategy

    在 Strategy 一文中。我已讲过Streategy的作用，用来Plan生成物理计划的。

这里提供了一种专门为了解析外部数据源的策略。

    最后会依据不同的BaseRelation生产不同的PhysicalRDD。

不同的BaseRelation的scan策略下文会介绍。

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()) :: Nilcase _ => 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进行Parse5、Parse后封装成为一个CreateTableUsing类的对象。

该类是一个RunnableCommand，其run方法会直接运行创建表语句。

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

7、得到external releation后，直接调用SQLContext的baseRelationToSchemaRDD转换为SchemaRDD8、最后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数据的规则。abstract class BaseRelation {def sqlContext: SQLContextdef schema: StructTypeabstract 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机制。在载入数据也的时候就进行过滤。而不是在client请求返回时做Filter。

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

RelationProvider：我们要实现这个，接受Parse后传入的參数。来生成相应的External Relation，就是一个反射生产外部数据源Relation的接口。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。
private[sql] case class JSONRelation(fileName: String, samplingRatio: Double)(@transient val sqlContext: SQLContext)extends TableScan {private def baseRDD = sqlContext.sparkContext.textFile(fileName) //读取json fileoverride 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、DefaultSourceparameters中能够获取到options中传入的path等自己定义參数。
这里接受传入的參数，来狗仔JsonRelation。

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)} }
六、总结External DataSource源代码分析下来。能够总结为3部分。

1、外部数据源的注冊流程2、外部数据源Table查询的计划解析流程3、怎样自己定义一个外部数据源，重写BaseRelation定义外部数据源的schema和scan的规则。定义RelationProvider。怎样生成外部数据源Relation。External Datasource此部分API还有可能在兴许的build中修改，眼下仅仅是涉及到了查询。关于其他的操作还未涉及。——EOF——

原创文章。转载请注明：

转载自：OopsOutOfMemory盛利的Blog。作者： OopsOutOfMemory

本文链接地址：http://blog.csdn.net/oopsoom/article/details/42064075  

注：本文基于署名-非商业性使用-禁止演绎 2.5 中国大陆(CC BY-NC-ND 2.5 CN)协议，欢迎转载、转发和评论，可是请保留本文作者署名和文章链接。如若须要用于商业目的或者与授权方面的协商，请联系我。

转载于:https://www.cnblogs.com/yangykaifa/p/6828063.html

总结

以上是生活随笔为你收集整理的Spark SQL之External DataSource外部数据源（二）源代码分析的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。