[SPARK-24005][CORE] Remove usage of Scala’s parallel collection #21913

MaxGekk · 2018-07-29T18:27:47Z

What changes were proposed in this pull request?

In the PR, I propose to replace Scala parallel collections by new methods parmap(). The methods use futures to transform a sequential collection by applying a lambda function to each element in parallel. The result of parmap is another regular (sequential) collection.

The proposed parmap method aims to solve the problem of impossibility to interrupt parallel Scala collection. This possibility is needed for reliable task preemption.

How was this patch tested?

A test was added to ThreadUtilsSuite

holdensmagicalunicorn · 2018-07-29T18:27:50Z

@MaxGekk, thanks! I am a bot who has found some folks who might be able to help with the review:@cloud-fan, @tdas and @rxin

SparkQA · 2018-07-29T18:34:33Z

Test build #93754 has finished for PR 21913 at commit 506fa93.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-07-29T22:32:13Z

Test build #93755 has finished for PR 21913 at commit 515d393.

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

SparkQA · 2018-07-30T02:43:40Z

Test build #93756 has finished for PR 21913 at commit e3df464.

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

SparkQA · 2018-07-30T13:30:05Z

Test build #93780 has finished for PR 21913 at commit 3601550.

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

gatorsmile · 2018-07-30T15:42:47Z

cc @zsxwing

srowen · 2018-07-31T18:28:39Z

What problem does this solve?

zsxwing · 2018-07-31T22:30:35Z

What problem does this solve?

@srowen readParquetFootersInParallel is called in executors. When a task is cancelled, it will still keeping run. If it reads lots of files, it will take pretty long and occupy the task slow. This change is basically to make it interruptible.

@MaxGekk Maybe add the following test to ThreadUtilsSuite. It shows what this PR is fixing.

  test("parmap should be interruptible") {
    val t = new Thread() {
      setDaemon(true)

      override def run() {
        try {
          // "par" is uninterruptible. The following will keep running even if the thread is
          // interrupted. We should prefer to use "ThreadUtils.parmap".
          //
          // (1 to 10).par.flatMap { i =>
          //   Thread.sleep(100000)
          //   1 to i
          // }
          //
          ThreadUtils.parmap(1 to 10, "test", 2) { i =>
            Thread.sleep(100000)
            1 to i
          }.flatten
        } catch {
          case _: InterruptedException => // excepted
        }
      }
    }
    t.start()
    eventually(timeout(10.seconds)) {
      assert(t.isAlive)
    }
    t.interrupt()
    eventually(timeout(10.seconds)) {
      assert(!t.isAlive)
    }
  }

zsxwing

Overall looks good. Left some comments.

zsxwing · 2018-07-31T22:31:18Z

streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLog.scala

-      val parallelCollection = group.par
-      parallelCollection.tasksupport = taskSupport
-      parallelCollection.map(handler)
+      ThreadUtils.parmap(group)(handler)(executionContext)


I would prefer to not change DStream codes.

Are there any specific reasons, or you think parmap is not reliable enough?

zsxwing · 2018-07-31T22:32:06Z

core/src/main/scala/org/apache/spark/util/ThreadUtils.scala

+   * @return new collection in which each element was given from the input collection `in` by
+   *         applying the lambda function `f`.
+   */
+  def parmap[I, O](


You can inline this method to the above one if you revert DStream changes.

zsxwing · 2018-07-31T22:33:47Z

core/src/main/scala/org/apache/spark/util/ThreadUtils.scala

+   *         applying the lambda function `f`.
+   */
+  def parmap[I, O](
+      in: TraversableOnce[I],


You can use Scala generic to make this method return the same Collection type as in. Such as

import scala.collection.TraversableLike import scala.collection.generic.CanBuildFrom import scala.language.higherKinds def parmap[I, O, Col[X] <: TraversableLike[X, Col[X]]] (in: Col[I], prefix: String, maxThreads: Int) (f: I => O) (implicit cbf: CanBuildFrom[Col[I], Future[O], Col[Future[O]]], // For in.map cbf2: CanBuildFrom[Col[Future[O]], O, Col[O]] // for Future.sequence ): Col[O] = { val pool = newForkJoinPool(prefix, maxThreads) try { implicit val ec = ExecutionContext.fromExecutor(pool) val futures = in.map(x => Future(f(x))) val futureSeq = Future.sequence(futures) awaitResult(futureSeq, Duration.Inf) } finally { pool.shutdownNow() } }

Then the caller side doesn't need to call toSeq.

@zsxwing Thank you very much for the code.

zsxwing · 2018-07-31T22:35:31Z

core/src/main/scala/org/apache/spark/util/ThreadUtils.scala

+      implicit val ec = ExecutionContext.fromExecutor(pool)
+      parmap(in)(f)
+    } finally {
+      pool.shutdown()


Use shutdownNow to interrupt the running tasks.

srowen · 2018-07-31T22:35:49Z

OK, got it. Yes that's worth mentioning in the doc too.
I'll have to go read up to get why this works differently, but obviously accept that it does.
There are other instances of .par, like in UnionRDD and in ddl.scala. Same treatment?

zsxwing · 2018-07-31T22:46:55Z

Yeah, let's also fix other instances.

SparkQA · 2018-08-03T15:00:59Z

Test build #94134 has finished for PR 21913 at commit 6a5f2ae.

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

srowen · 2018-08-03T15:17:20Z

core/src/main/scala/org/apache/spark/util/ThreadUtils.scala

  }
+
+  /**
+   * Transforms input collection by applying the given function to each element in parallel fashion.


I'd still include a note in these docs about what this does differently from .par. Just a sentence about it being interruptible.

I added a comment about this.

SparkQA · 2018-08-05T15:18:36Z

Test build #94242 has finished for PR 21913 at commit cf48d8d.

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

MaxGekk · 2018-08-05T17:02:40Z

jenkins, retest this, please

SparkQA · 2018-08-05T18:27:30Z

Test build #94246 has finished for PR 21913 at commit cf48d8d.

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

SparkQA · 2018-08-05T23:23:35Z

Test build #94248 has finished for PR 21913 at commit 610154c.

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

cloud-fan · 2018-08-07T09:15:02Z

LGTM, merging to master!

ConeyLiu · 2019-02-19T13:34:28Z

core/src/test/scala/org/apache/spark/util/ThreadUtilsSuite.scala

+    eventually(timeout(10.seconds)) {
+      assert(!t.isAlive)
+    }
+  }


Hi, @MaxGekk @zsxwing Could you tell me why this can be interrupted? while (1 to 10).par can't.

zsxwing · 2019-02-19T18:22:36Z

core/src/main/scala/org/apache/spark/util/ThreadUtils.scala

+
+      parmap(in)(f)
+    } finally {
+      pool.shutdownNow()


@ConeyLiu this line interrupts the tasks in the thread pool. Scala par doesn't do this.

@zsxwing, thanks very much for your answer.

MaxGekk added 2 commits July 29, 2018 19:15

Initial implementation of parmap function

ac26dea

Initial implementation of parmap()

506fa93

Making Scala style checker happy

515d393

Fix timeout

e3df464

Fix test - expect SparkException

3601550

zsxwing requested changes Jul 31, 2018

View reviewed changes

zsxwing reviewed Jul 31, 2018

View reviewed changes

MaxGekk added 6 commits August 3, 2018 11:59

Merge remote-tracking branch 'origin/master' into par-map

c711d80

Adding a test provided by Shixiong Zhu

592b5a7

shutdown -> shutdownNow

a863bd4

Restricting type parameters of parmap

ad03004

Porting UnionRDD on parmap

72cdfeb

Porting ALTER TABLE on parmap

6a5f2ae

MaxGekk changed the title ~~[WIP][SPARK-24005][CORE] Remove usage of Scala’s parallel collection~~ [SPARK-24005][CORE] Remove usage of Scala’s parallel collection Aug 3, 2018

srowen reviewed Aug 3, 2018

View reviewed changes

MaxGekk added 2 commits August 5, 2018 14:58

Merge remote-tracking branch 'origin/master' into par-map

701901c

Improving comments

cf48d8d

Bug fix: recursive calls

610154c

asfgit closed this in 131ca14 Aug 7, 2018

ConeyLiu reviewed Feb 19, 2019

View reviewed changes

zsxwing reviewed Feb 19, 2019

View reviewed changes

MaxGekk deleted the par-map branch August 17, 2019 13:33

MaxGekk mentioned this pull request Oct 9, 2019

[SPARK-29413][CORE] Rewrite ThreadUtils.parmap to avoid TraversableLike, gone in 2.13 #26072

Closed

[SPARK-24005][CORE] Remove usage of Scala’s parallel collection #21913

[SPARK-24005][CORE] Remove usage of Scala’s parallel collection #21913

Uh oh!

Conversation

MaxGekk commented Jul 29, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

holdensmagicalunicorn commented Jul 29, 2018

Uh oh!

SparkQA commented Jul 29, 2018

Uh oh!

SparkQA commented Jul 29, 2018

Uh oh!

SparkQA commented Jul 30, 2018

Uh oh!

SparkQA commented Jul 30, 2018

Uh oh!

gatorsmile commented Jul 30, 2018

Uh oh!

srowen commented Jul 31, 2018

Uh oh!

zsxwing commented Jul 31, 2018

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing Jul 31, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srowen commented Jul 31, 2018

Uh oh!

zsxwing commented Jul 31, 2018

Uh oh!

SparkQA commented Aug 3, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 5, 2018

Uh oh!

MaxGekk commented Aug 5, 2018

Uh oh!

SparkQA commented Aug 5, 2018

Uh oh!

SparkQA commented Aug 5, 2018

Uh oh!

cloud-fan commented Aug 7, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

MaxGekk commented Jul 29, 2018 •

edited

Loading

zsxwing Jul 31, 2018 •

edited

Loading