[SPARK-19443][SQL] The function to generate constraints takes too long when the query plan grows continuously

[viirya]

What changes were proposed in this pull request?

This issue is originally reported and discussed at http://apache-spark-developers-list.1001551.n3.nabble.com/SQL-ML-Pipeline-performance-regression-between-1-6-and-2-x-tc20803.html

When run a ML Pipeline with many stages, during the iterative updating to Dataset , it is observed the it takes longer time to finish the fit and transform as the query plan grows continuously.

The example code show as the following in benchmark.

Specially, the time spent on preparing optimized plan in current branch is much higher than 1.6. Actually, the time is spent mostly on generating query plan's constraints during few optimization rules.

getAliasedConstraints is found to be a function costing most of the running time. As the constraints for aliasing will increase very fast.

This patch tries to improve the performance of getAliasedConstraints.

Benchmark

Run the following codes locally.

import org.apache.spark.ml.{Pipeline, PipelineStage}
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer, VectorAssembler}

val df = (1 to 40).foldLeft(Seq((1, "foo"), (2, "bar"), (3, "baz")).toDF("id", "x0"))((df, i) => df.withColumn(s"x$i", $"x0"))

val indexers = df.columns.tail.map(c => new StringIndexer()
  .setInputCol(c)
  .setOutputCol(s"${c}_indexed")
  .setHandleInvalid("skip"))

val encoders = indexers.map(indexer => new OneHotEncoder()
  .setInputCol(indexer.getOutputCol)
  .setOutputCol(s"${indexer.getOutputCol}_encoded")
  .setDropLast(true))

val stages: Array[PipelineStage] = indexers ++ encoders
val pipeline = new Pipeline().setStages(stages)

val startTime = System.nanoTime
pipeline.fit(df).transform(df).show
val runningTime = System.nanoTime - startTime

Before this patch: 1786001 ms
After this patch: 843688 ms

More than half of original running time is saved.

How was this patch tested?

Jenkins tests.

Please review http://spark.apache.org/contributing.html before opening a pull request.

[SparkQA]

Test build #72303 has started for PR 16785 at commit b4e514a.

[viirya]

retest this please.

[SparkQA]

Test build #72305 has finished for PR 16785 at commit b4e514a.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

The rewritten logic is not correct. I am working to improve this with other approach.

[viirya]

I don't find a way to improve getAliasedConstraints significantly by re-writing its logic. The current way to improve its performance is using parallel collection to do the transformation in parallel. It can cut the running time by half (see benchmark in the pr description), but the running time (13.5 secs) is still too long compared with 1.6.

We may consider #16775 which is an another solution to fix this issue by checkpointing datasets for pipelines of long stages, or both of them.

[viirya]

since this change is related to SQL, cc @cloud-fan @hvanhovell

[SparkQA]

Test build #72625 has finished for PR 16785 at commit 8c98a5c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[mallman]

Why are we using a custom task support instead of the default (which uses the global fork-join executor)?

[viirya]

Do they have the same parallelism level? BTW, I saw the parallel collection used in other places in Spark all take custom task support.

[mallman]

Whether they do or not depends on the implementation of the default task support. But even if they use the same level of parallelism, they're distinct executors. Which means they won't share a common thread pool or task queue. I don't know why Spark would use custom task support in other places. It may be to avoid engaging all of the CPU cores on the host machine. But then it seems more efficient for Spark to have its own global task support.

[hvanhovell]

@viirya this looks like a very big hammer to solve this problem. Can't we try a different approach?

I think we should try to avoid optimizing already optimized code snippets, you might be able to do this using some kind of a fence. It would even be better if we would have a recursive node.

[viirya]

@hvanhovell Yeah, I think so. As in previous comment, I don't find a way now to improve getAliasedConstraints significantly by re-writing its logic without parallel collection.

We may consider #16775 which is an another solution to fix this issue by checkpointing datasets for pipelines of long stages.

[cloud-fan]

can we consider this in a higher-level of view instead of focusing on the method getAliasedConstraints? Maybe there is a way to do constraint propagation faster.

[viirya]

@cloud-fan yeah, i agreed with you and @hvanhovell.

For too slow constraint propagation, except for attacking getAliasedConstraints like this change, maybe we can have other way to improve the process doing constraint propagation.

If we can't, for such long lineages, I think we should use checkpointing to fix it like #16775.

[viirya]

this looks like a very big hammer to solve this problem. Can't we try a different approach?
I think we should try to avoid optimizing already optimized code snippets, you might be able to do this using some kind of a fence. It would even be better if we would have a recursive node.

@cloud-fan @hvanhovell Ok. I've figured out to add a filter to reduce the candidates of aliased constraints. It can achieve same speed-up (cut of half running time in benchmark) without parallel collection hammer.

Can you have time to look at it? Thanks.

[SparkQA]

Test build #73034 has finished for PR 16785 at commit 4ba93fe.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #73035 has finished for PR 16785 at commit 278c31c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[hvanhovell]

cc @sameeragarwal

[sameeragarwal]

Hi @viirya, I understand that #16998 probably supersedes this, but just out of curiosity, did you see a lot of benefit from pruning these attributes here given that we already prune them later in QueryPlan?

[viirya]

Yes. You can see the benchmark in the pr description. With pruning these attributes, the running time is cut half.

If I understand your comment correctly, pruning them later in QueryPlan means we prune constraints which don't refer attributes in outputSet.

But the pruning here is happened before the pruning you pointed out, we need to reduce the constraints taken for transforming aliasing attributes to lower the computation cost.