结构化处理入门-SparkSession

建立DataFrame，有三种模式，一种是sql()主要是访问Hive表；一种是从RDD生成DataFrame，主要从ExistingRDD开始建立；还有一种是read/format格式，从json/txt/csv等数据源格式建立。

先看看第三种方式的建立流程。

一、read/format

def read: DataFrameReader = new DataFrameReader(self)

SparkSession.read()方法直接建立DataFrameReader，而后再DataFrameReader的load()方法来导入外部数据源。load()方法主要逻辑以下：
 

def load(paths: String*): DataFrame = {

    sparkSession.baseRelationToDataFrame(

      DataSource.apply(

        sparkSession,

        paths = paths,

        userSpecifiedSchema = userSpecifiedSchema,

        className = source,

        options = extraOptions.toMap).resolveRelation())

  }

建立对应数据源类型的DataSource，DataSource解析成BaseRelation,而后经过SparkSession的baseRelationToDataFrame方法从BaseRelation映射生成DataFrame。从BaseRelation建立LogicalRelation，而后调用Dataset.ofRows方法从LogicalRelation建立DataFrame。DataFrame实际就是Dataset。

type DataFrame = Dataset[Row]

baseRelationToDataFrame的定义：

def baseRelationToDataFrame(baseRelation: BaseRelation): DataFrame = {

    Dataset.ofRows(self, LogicalRelation(baseRelation))

  }

Dataset.ofRows方法主要是将逻辑计划转换成物理计划，而后生成新的Dataset。

二、执行

SparkSession的执行关键是如何从LogicalPlan生成物理计划。咱们试试跟踪这部分逻辑。

def count(): Long = withAction("count", groupBy().count().queryExecution) { plan =>

    plan.executeCollect().head.getLong(0)

  }

Dataset的count()动做触发物理计划的执行，调用物理计划plan的executeCollect方法，该方法实际上会调用doExecute()方法生成Array[InternalRow]格式。executeCollect方法在SparkPlan中定义。

三、HadoopFsRelation

须要跟踪下如何从HadoopFsRelation生成物理计划（也就是SparkPlan）

经过FileSourceStrategy来解析。它在FileSourceScanExec上叠加Filter和Projection等操做，看看FileSourceScanExec的定义：

case class FileSourceScanExec(

    @transient relation: HadoopFsRelation,

    output: Seq[Attribute],

    requiredSchema: StructType,

    partitionFilters: Seq[Expression],

    dataFilters: Seq[Expression],

    override val metastoreTableIdentifier: Option[TableIdentifier])

  extends DataSourceScanExec with ColumnarBatchScan  {

。。。

}

它的主要执行代码doExecute()的功能逻辑以下：

protected override def doExecute(): RDD[InternalRow] = {

    if (supportsBatch) {

      // in the case of fallback, this batched scan should never fail because of:

      // 1) only primitive types are supported

      // 2) the number of columns should be smaller than spark.sql.codegen.maxFields

      WholeStageCodegenExec(this).execute()

    } else {

      val unsafeRows = {

        val scan = inputRDD

        if (needsUnsafeRowConversion) {

          scan.mapPartitionsWithIndexInternal { (index, iter) =>

            val proj = UnsafeProjection.create(schema)

            proj.initialize(index)

            iter.map(proj)

          }

        } else {

          scan

        }

      }

      val numOutputRows = longMetric("numOutputRows")

      unsafeRows.map { r =>

        numOutputRows += 1

        r

      }

    }

  }

inputRDD有两种方式建立，一是createBucketedReadRDD，二是createNonBucketedReadRDD。二者没有本质的区别，仅仅是文件分区规则的不一样。

private lazy val inputRDD: RDD[InternalRow] = {

    val readFile: (PartitionedFile) => Iterator[InternalRow] =

      relation.fileFormat.buildReaderWithPartitionValues(

        sparkSession = relation.sparkSession,

        dataSchema = relation.dataSchema,

        partitionSchema = relation.partitionSchema,

        requiredSchema = requiredSchema,

        filters = pushedDownFilters,

        options = relation.options,

        hadoopConf = relation.sparkSession.sessionState.newHadoopConfWithOptions(relation.options))



    relation.bucketSpec match {

      case Some(bucketing) if relation.sparkSession.sessionState.conf.bucketingEnabled =>

        createBucketedReadRDD(bucketing, readFile, selectedPartitions, relation)

      case _ =>

        createNonBucketedReadRDD(readFile, selectedPartitions, relation)

    }

  }

createNonBucketedReadRDD调用FileScanRDD ：

new FileScanRDD(fsRelation.sparkSession, readFile, partitions)