官术网_书友最值得收藏!

Predicate push-down on smart data sources

Smart data sources are those that support data processing directly in their own engine--where the data resides--by preventing unnecessary data to be sent to Apache Spark.

On example is a relational SQL database with a smart data source. Consider a table with three columns: column1, column2, and column3, where the third column contains a timestamp. In addition, consider an ApacheSparkSQL query using this JDBC data source but only accessing a subset of columns and rows based using projection and selection. The following SQL query is an example of such a task:

select column2,column3 from tab where column3>1418812500

Running on a smart data source, data locality is made use of by letting the SQL database do the filtering of rows based on timestamp and removal of column1.

Let's have a look at a practical example on how this is implemented in the Apache Spark MongoDB connector. First, we'll take a look at the class definition:

private[spark] case class MongoRelation(mongoRDD: MongoRDD[BsonDocument], _schema: Option[StructType])(@transient val sqlContext: SQLContext)
extends BaseRelation
with PrunedFilteredScan
with InsertableRelation
with LoggingTrait {

As you can see, the MongoRelation class extends BaseRelation. This is all that is needed to create a new plugin to the DataSource API in order to support an additional data source. However, this class also implemented the PrunedFilteredScan trait adding the buildScan method in order to support filtering on the data source itself. So let's take a look at the implementation of this method:

override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
// Fields that explicitly aren't nullable must also be added to the filters
val pipelineFilters = schema
.fields
.filter(!_.nullable)
.map(_.name)
.map(IsNotNull)
++ filters

if (requiredColumns.nonEmpty || pipelineFilters.nonEmpty) {
logInfo(s"requiredColumns: ${requiredColumns.mkString(", ")},
filters: ${pipelineFilters.mkString(", ")}")
}
mongoRDD.appendPipeline(createPipeline(requiredColumns, pipelineFilters))
.map(doc => documentToRow(doc, schema, requiredColumns))
}

It is not necessary to understand the complete code snippet, but you can see that two parameters are passed to the buildScan method: requiredColumns and filters. This means that the code can use this information to remove columns and rows directly using the MongoDB API.

主站蜘蛛池模板: 遵化市| 湾仔区| 东台市| 泾川县| 科尔| 丹凤县| 青河县| 齐河县| 德惠市| 禄丰县| 武平县| 清苑县| 郧西县| 盘山县| 武山县| 恩平市| 攀枝花市| 苏尼特左旗| 黑河市| 长沙县| 汉源县| 富顺县| 宜良县| 崇礼县| 衡水市| 桐梓县| 聂拉木县| 高碑店市| 葵青区| 西安市| 八宿县| 五台县| 龙井市| 宜昌市| 阳江市| 随州市| 色达县| 河南省| 内黄县| 济源市| 宁强县|