[SPARK-19486][CORE](try 3) Investigate using multiple threads for task serialization

[witgo]

What changes were proposed in this pull request?

See https://issues.apache.org/jira/browse/SPARK-19486

In the case of stage has a lot of tasks, this PR can improve the scheduling performance of 15%

The test code:

val rdd = sc.parallelize(0 until 100).repartition(100000)
rdd.localCheckpoint().count()
rdd.sum()
(1 to 10).foreach{ i=>
  val serializeStart = System.nanoTime()
  rdd.sum()
  val serializeFinish = System.nanoTime()
  println(f"${(serializeFinish - serializeStart) / 1E9}%1.4f")
}

and spark-defaults.conf file:

spark.master                                      yarn-client
spark.executor.instances                          20
spark.driver.memory                               64g
spark.executor.memory                             30g
spark.executor.cores                              5
spark.default.parallelism                         100 
spark.sql.shuffle.partitions                      100
spark.serializer                                  org.apache.spark.serializer.KryoSerializer
spark.driver.maxResultSize                        0
spark.ui.enabled                                  false 
spark.driver.extraJavaOptions                     -XX:+UseG1GC -XX:+UseStringDeduplication -XX:G1HeapRegionSize=16M -XX:MetaspaceSize=512M 
spark.executor.extraJavaOptions                   -XX:+UseG1GC -XX:+UseStringDeduplication -XX:G1HeapRegionSize=16M -XX:MetaspaceSize=256M 
spark.cleaner.referenceTracking.blocking          true
spark.cleaner.referenceTracking.blocking.shuffle  true

The test results are as follows

partition	SPARK-18890	db0ddce
100	0.0273 s	0.028 s
1K	0.1234 s	0.1321 s
10k	0.6557 s	0.9502 s
100K	6.1541 s	9.4179 s

How was this patch tested?

Existing tests.

[SparkQA]

Test build #73775 has finished for PR 17139 at commit a06f8c8.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SerializeTask(task: TaskDescription) extends CoarseGrainedClusterMessage
• class SerializeTaskEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint with Logging

[SparkQA]

Test build #73776 has finished for PR 17139 at commit 6874d1e.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SerializeTask(task: TaskDescription) extends CoarseGrainedClusterMessage
• class SerializeTaskEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint with Logging

[SparkQA]

Test build #73769 has finished for PR 17139 at commit bfa285b.

• This patch fails from timeout after a configured wait of `250m`.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SerializeTask(task: TaskDescription) extends CoarseGrainedClusterMessage
• class SerializeTaskEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint with Logging

[SparkQA]

Test build #73771 has finished for PR 17139 at commit af5fc9f.

• This patch fails from timeout after a configured wait of `250m`.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SerializeTask(task: TaskDescription) extends CoarseGrainedClusterMessage
• class SerializeTaskEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint with Logging

[SparkQA]

Test build #73848 has finished for PR 17139 at commit 4d9b666.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class SerializeTask(task: TaskDescription) extends CoarseGrainedClusterMessage
• class SerializeTaskEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint with Logging

[witgo]

ping @kayousterhout @squito

[kayousterhout]

Why is the time improvement so much larger here than in the other PR?

[witgo]

Added the multi-threaded code for serialization TaskDescription .

[kayousterhout]

Can you also post the time differences for some smaller jobs (e.g., 100 tasks, 1000 tasks, 10K tasks) to get a sense of how this varies with size?

[witgo]

@kayousterhout The test report has been updated.

[SparkQA]

Test build #74270 has finished for PR 17139 at commit 8fbe15c.

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

[jiangxb1987]

@witgo Are you still working on this?

[witgo]

@jiangxb1987 ,Yes do you have any questions?

[jiangxb1987]

Please bring this PR up to date, then maybe some guy can review this. :)

[srowen]

This should probably just be closed

[gatorsmile]

We are closing it due to inactivity. please do reopen if you want to push it forward. Thanks!