Spark连接外部数据源解读

时间:2023-01-03 19:06:53

本文以连接HBase数据库为例,介绍Spark DataSource API的结构。项目源码:https://github.com/hortonworks-spark/shc 注:由于某些原因,尚无充裕时间进行更深入的解读,本文先着重数据源注册和写入两个流程,后续文章会继续跟进。本文原文出处: http://blog.csdn.net/bluishglc/article/details/52882250 严禁任何形式的转载,否则将委托CSDN官方维护权益!

注册流程

位置:org.apache.spark.sql.execution.datasources.hbase.examples.HBaseSource,Line:75

def withCatalog(cat: String): DataFrame = {
sqlContext
.read
.options(Map(HBaseTableCatalog.tableCatalog->cat))
.format("org.apache.spark.sql.execution.datasources.hbase")
.load()
}

read方法会创建一个DataFrameReader实例并返回,后续的options和format是在为新创建的DataFrameReader实例注册相关信息,最终是通过load方法,在基于options和format传入的信息基础之上,初始化了一个DataFrame.

位置: org.apache.spark.sql.DataFrameReader#load, Line: 118

  def load(): DataFrame = {
val resolved = ResolvedDataSource(
sqlContext,
userSpecifiedSchema = userSpecifiedSchema,
partitionColumns = Array.empty[String],
provider = source,
options = extraOptions.toMap)
DataFrame(sqlContext, LogicalRelation(resolved.relation))
}

在创建DataFrame实例时,最重要的是需要传入一个LogicalRelation, 而LogicalRelation需要一个BaseRelation(对BaseRelation进行包裹),BaseRelation的主要用途是描述数据的Schema,这里的我们要传入的BaseRelation实例是通过ResolvedDataSource的apply方法创建的字段:relation, 这个relation的创建发生在:org.apache.spark.sql.execution.datasources.ResolvedDataSource#apply Line 153 - 158

      case None => clazz.newInstance() match {
case dataSource: RelationProvider =>
val caseInsensitiveOptions = new CaseInsensitiveMap(options)
if (caseInsensitiveOptions.contains("paths")) {
throw new AnalysisException(s"$className does not support paths option.")
}
dataSource.createRelation(sqlContext, caseInsensitiveOptions)

最终,这个Relation是通过org.apache.spark.sql.execution.datasources.hbase.DefaultSource#createRelation创建的(DefaultSource有两个createRelation方法,一个是for read, 一个是for write),而这个Relation的实例是org.apache.spark.sql.execution.datasources.hbase.HBaseRelation.

到这里,我们可以简单小结一下:在Spark中注册和读取一个新的数据源,只需要我们给出数据源的Class路径和一些需要的参数,就像本例中options和format那样做就可以了。我们真正需要做的工作是提供RelationProvider和对应的Relation。

写入流程

位置:org.apache.spark.sql.execution.datasources.hbase.examples.HBaseSource#main Line: 85 - 91

val data = (0 to 255).map { i =>
HBaseRecord(i)
}
sc.parallelize(data).toDF.write.options(
Map(HBaseTableCatalog.tableCatalog -> cat, HBaseTableCatalog.newTable -> "5"))
.format("org.apache.spark.sql.execution.datasources.hbase")
.save()

有几处需要注意的地方:
- sc.parallelize(data)返回的是一个RDD,而toDF则是org.apache.spark.sql.DataFrameHolder的一个方法,很显然,这里发生了隐式转换。隐士转化是声明是发生在前面的一行代码里import sqlContext.implicits._ implicits是SQLContext类内部定义的一个object, 它本身并无过多的隐式转换定义,大量的隐式转换都定义在了它的父类org.apache.spark.sql.SQLImplicits里面,而我们上面的代码使用到的隐式转换正是org.apache.spark.sql.SQLImplicits#rddToDataFrameHolder,具体如下

implicit def rddToDataFrameHolder[A <: Product : TypeTag](rdd: RDD[A]): DataFrameHolder = {
DataFrameHolder(_sqlContext.createDataFrame(rdd))
}

它将一个RDD隐式转化成为了DataFrameHolder,进而才能调用toDF方法去得到一个DataFrame。

从这个方法中我们可以看到这个DataFrame的创建最最终还是通过org.apache.spark.sql.SQLContext#createDataFrame方法实现的。看上去有些“绕”,这可能是版本在演化过程中的历史原因造成的。我们来具体看一下这个方法:

  def createDataFrame[A <: Product : TypeTag](rdd: RDD[A]): DataFrame = {
SQLContext.setActive(self)
val schema = ScalaReflection.schemaFor[A].dataType.asInstanceOf[StructType]
val attributeSeq = schema.toAttributes
val rowRDD = RDDConversions.productToRowRdd(rdd, schema.map(_.dataType))
DataFrame(self, LogicalRDD(attributeSeq, rowRDD)(self))
}

对于这个方法的一些细节我们先不做过多深入的探究,简单地总结起来就是:SQLContext提供了把RDD转换成DataFrame的方法,你只需要提供数据的Schema信息即可,如果你没有显示的提供Schema,SQLContext在进行parallelize操作把数据transform成RDD时会根据传入的数据类型参数[T]以反射的方式获取Schema信息,在本例中传入的数据类型是:org.apache.spark.sql.execution.datasources.hbase.examples.HBaseRecord

DataFrame实例创建出出来之后,通过write方法创建了一个org.apache.spark.sql.DataFrameWriter实例,然后通过options和format为新创建的DataFrameWriter实例注册相关信息,最后通过save方法将DataFrame实例中的数据时保存起来。实际的保存动作是发生是依赖org.apache.spark.sql.execution.datasources.ResolvedDataSource#apply完成的。注意: ResolvedDataSource有两个重载的apply的方法,一个为读取服务的(这个方法前文已经提及)一个是为写入服务的,我们具体看一下这个为写入提供的apply方法:

    val relation = clazz.newInstance() match {
case dataSource: CreatableRelationProvider =>
dataSource.createRelation(sqlContext, mode, options, data)
......
}
ResolvedDataSource(clazz, relation)
}

和读取的流程很类似,这个apply方法也要创建一个Relation,这个Relation是通过org.apache.spark.sql.execution.datasources.hbase.DefaultSource#createRelation创建的(DefaultSource有两个createRelation方法,一个是for read, 一个是for write),让我们来看一下这个实际执行写入的方法:

  override def createRelation(
sqlContext: SQLContext,
mode: SaveMode,
parameters: Map[String, String],
data: DataFrame): BaseRelation = {
val relation = HBaseRelation(parameters, Some(data.schema))(sqlContext)
relation.createTable()
relation.insert(data, false)
relation
}

在创建Relation的过程中,代码附带性的创建了Table,插入了数据。这一部分的设计不是很好,方法的目的不够单一,应该相应的改动一下。