Repository: spark Updated Branches: refs/heads/master c9ba59d38 -> 34f229bc2
[SPARK-25710][SQL] range should report metrics correctly ## What changes were proposed in this pull request? Currently `Range` reports metrics in batch granularity. This is acceptable, but it's better if we can make it row granularity without performance penalty. Before this PR, the metrics are updated when preparing the batch, which is before we actually consume data. In this PR, the metrics are updated after the data are consumed. There are 2 different cases: 1. The data processing loop has a stop check. The metrics are updated when we need to stop. 2. no stop check. The metrics are updated after the loop. ## How was this patch tested? existing tests and a new benchmark Closes #22698 from cloud-fan/range. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/34f229bc Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/34f229bc Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/34f229bc Branch: refs/heads/master Commit: 34f229bc2116e443fa7f4e20d5277c4a27d02c47 Parents: c9ba59d Author: Wenchen Fan <[email protected]> Authored: Sat Oct 13 13:55:28 2018 +0800 Committer: Wenchen Fan <[email protected]> Committed: Sat Oct 13 13:55:28 2018 +0800 ---------------------------------------------------------------------- sql/core/benchmarks/RangeBenchmark-results.txt | 16 +++++ .../sql/execution/basicPhysicalOperators.scala | 17 +++-- .../execution/benchmark/RangeBenchmark.scala | 65 ++++++++++++++++++++ .../sql/execution/metric/SQLMetricsSuite.scala | 10 +-- 4 files changed, 98 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/spark/blob/34f229bc/sql/core/benchmarks/RangeBenchmark-results.txt ---------------------------------------------------------------------- diff --git a/sql/core/benchmarks/RangeBenchmark-results.txt b/sql/core/benchmarks/RangeBenchmark-results.txt new file mode 100644 index 0000000..21766e0 --- /dev/null +++ b/sql/core/benchmarks/RangeBenchmark-results.txt @@ -0,0 +1,16 @@ +================================================================================================ +range +================================================================================================ + +Java HotSpot(TM) 64-Bit Server VM 1.8.0_161-b12 on Mac OS X 10.13.6 +Intel(R) Core(TM) i7-6920HQ CPU @ 2.90GHz + +range: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------ +full scan 12674 / 12840 41.4 24.2 1.0X +limit after range 33 / 37 15900.2 0.1 384.4X +filter after range 969 / 985 541.0 1.8 13.1X +count after range 42 / 42 12510.5 0.1 302.4X +count after limit after range 32 / 33 16337.0 0.1 394.9X + + http://git-wip-us.apache.org/repos/asf/spark/blob/34f229bc/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala index 4cd2e78..09effe0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala @@ -452,8 +452,15 @@ case class RangeExec(range: org.apache.spark.sql.catalyst.plans.logical.Range) val localIdx = ctx.freshName("localIdx") val localEnd = ctx.freshName("localEnd") - val shouldStop = if (parent.needStopCheck) { - s"if (shouldStop()) { $nextIndex = $value + ${step}L; return; }" + val stopCheck = if (parent.needStopCheck) { + s""" + |if (shouldStop()) { + | $nextIndex = $value + ${step}L; + | $numOutput.add($localIdx + 1); + | $inputMetrics.incRecordsRead($localIdx + 1); + | return; + |} + """.stripMargin } else { "// shouldStop check is eliminated" } @@ -506,8 +513,6 @@ case class RangeExec(range: org.apache.spark.sql.catalyst.plans.logical.Range) | $numElementsTodo = 0; | if ($nextBatchTodo == 0) break; | } - | $numOutput.add($nextBatchTodo); - | $inputMetrics.incRecordsRead($nextBatchTodo); | $batchEnd += $nextBatchTodo * ${step}L; | } | @@ -515,9 +520,11 @@ case class RangeExec(range: org.apache.spark.sql.catalyst.plans.logical.Range) | for (int $localIdx = 0; $localIdx < $localEnd; $localIdx++) { | long $value = ((long)$localIdx * ${step}L) + $nextIndex; | ${consume(ctx, Seq(ev))} - | $shouldStop + | $stopCheck | } | $nextIndex = $batchEnd; + | $numOutput.add($localEnd); + | $inputMetrics.incRecordsRead($localEnd); | $taskContext.killTaskIfInterrupted(); | } """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/34f229bc/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/RangeBenchmark.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/RangeBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/RangeBenchmark.scala new file mode 100644 index 0000000..a844e02 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/RangeBenchmark.scala @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.benchmark + +import org.apache.spark.benchmark.Benchmark + +/** + * Benchmark to measure performance for range operator. + * To run this benchmark: + * {{{ + * 1. without sbt: + * bin/spark-submit --class <this class> --jars <spark core test jar> <spark sql test jar> + * 2. build/sbt "sql/test:runMain <this class>" + * 3. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain <this class>" + * Results will be written to "benchmarks/RangeBenchmark-results.txt". + * }}} + */ +object RangeBenchmark extends SqlBasedBenchmark { + + override def runBenchmarkSuite(): Unit = { + import spark.implicits._ + + runBenchmark("range") { + val N = 500L << 20 + val benchmark = new Benchmark("range", N, output = output) + + benchmark.addCase("full scan", numIters = 4) { _ => + spark.range(N).queryExecution.toRdd.foreach(_ => ()) + } + + benchmark.addCase("limit after range", numIters = 4) { _ => + spark.range(N).limit(100).queryExecution.toRdd.foreach(_ => ()) + } + + benchmark.addCase("filter after range", numIters = 4) { _ => + spark.range(N).filter('id % 100 === 0).queryExecution.toRdd.foreach(_ => ()) + } + + benchmark.addCase("count after range", numIters = 4) { _ => + spark.range(N).count() + } + + benchmark.addCase("count after limit after range", numIters = 4) { _ => + spark.range(N).limit(100).count() + } + + benchmark.run() + } + } +} http://git-wip-us.apache.org/repos/asf/spark/blob/34f229bc/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala index 81db3e1..b955c15 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala @@ -556,16 +556,16 @@ class SQLMetricsSuite extends SparkFunSuite with SQLMetricsTestUtils with Shared df.queryExecution.executedPlan.foreach(_.resetMetrics()) // For each partition, we get 2 rows. Then the Filter should produce 2 rows per-partition, - // and Range should produce 1000 rows (one batch) per-partition. Totally Filter produces - // 4 rows, and Range produces 2000 rows. + // and Range should produce 4 rows per-partition ([0, 1, 2, 3] and [15, 16, 17, 18]). Totally + // Filter produces 4 rows, and Range produces 8 rows. df.queryExecution.toRdd.mapPartitions(_.take(2)).collect() - checkFilterAndRangeMetrics(df, filterNumOutputs = 4, rangeNumOutputs = 2000) + checkFilterAndRangeMetrics(df, filterNumOutputs = 4, rangeNumOutputs = 8) // Top-most limit will call `CollectLimitExec.executeCollect`, which will only run the first - // task, so totally the Filter produces 2 rows, and Range produces 1000 rows (one batch). + // task, so totally the Filter produces 2 rows, and Range produces 4 rows ([0, 1, 2, 3]). val df2 = df.limit(2) df2.collect() - checkFilterAndRangeMetrics(df2, filterNumOutputs = 2, rangeNumOutputs = 1000) + checkFilterAndRangeMetrics(df2, filterNumOutputs = 2, rangeNumOutputs = 4) } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
