【原創】大叔經驗分享（15）spark sql limit實現原理

之前討論過hive中limit的實現，詳見 https://www.cnblogs.com/barneywill/p/10109217.html
下面看spark sql中limit的實現，首先看執行計划：

spark-sql> explain select * from test1 limit 10;
== Physical Plan ==
CollectLimit 10
+- HiveTableScan [id#35], MetastoreRelation temp, test1
Time taken: 0.201 seconds, Fetched 1 row(s)

limit對應的CollectLimit，對應的實現類是

org.apache.spark.sql.execution.CollectLimitExec

case class CollectLimitExec(limit: Int, child: SparkPlan) extends UnaryExecNode {
...
  protected override def doExecute(): RDD[InternalRow] = {
    val locallyLimited = child.execute().mapPartitionsInternal(_.take(limit))
    val shuffled = new ShuffledRowRDD(
      ShuffleExchange.prepareShuffleDependency(
        locallyLimited, child.output, SinglePartition, serializer))
    shuffled.mapPartitionsInternal(_.take(limit))
  }

可見實現非常簡單，首先調用SparkPlan.execute得到結果的RDD，然后從每個partition中取前limit個row得到一個新的RDD，然后再將這個新的RDD變成一個分區，然后再取前limit個，這樣就得到最終的結果。