[SPARK-19991]FileSegmentManagedBuffer performance improvement

[witgo]

FileSegmentManagedBuffer performance improvement.

What changes were proposed in this pull request?

When we do not set the value of the configuration items spark.storage.memoryMapThreshold and spark.shuffle.io.lazyFD,
each call to the cFileSegmentManagedBuffer.nioByteBuffer or FileSegmentManagedBuffer.createInputStream method creates a NoSuchElementException instance. This is a more time-consuming operation.

In the use case, this PR can improve the performance of about 3.5%

The test code:

(1 to 10).foreach { i =>
  val numPartition = 10000
  val rdd = sc.parallelize(0 until numPartition).repartition(numPartition).flatMap { t =>
    (0 until numPartition).map(r => r * numPartition + t)
  }.repartition(numPartition)
  val serializeStart = System.currentTimeMillis()
  rdd.sum()
  val serializeFinish = System.currentTimeMillis()
  println(f"Test $i: ${(serializeFinish - serializeStart) / 1000D}%1.2f")
}

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

SPARK-19991	68ea290
226.09 s	235.21 s

How was this patch tested?

Existing tests.

[SparkQA]

Test build #74714 has finished for PR 17329 at commit abcfc79.

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

[SparkQA]

Test build #74715 has finished for PR 17329 at commit dc5fd8d.

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

[SparkQA]

Test build #74727 has started for PR 17329 at commit e0a3b62.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/74727/
Test FAILed.

[srowen]

Out of curiosity why does each call generate NoSuchElementException ?

[witgo]

public class HadoopConfigProvider extends ConfigProvider {
  private final Configuration conf;

  public HadoopConfigProvider(Configuration conf) {
    this.conf = conf;
  }

  @Override
  public String get(String name) {
    String value = conf.get(name);
    // When do not set the value of spark.storage.memoryMapThreshold or spark.shuffle.io.lazyFD,
   //  the value of `value` is null
    if (value == null) {
      throw new NoSuchElementException(name);
    }
    return value;
  }

  @Override
  public Iterable<Map.Entry<String, String>> getAll() {
    return conf;
  }

}

[srowen]

Yeah that makes sense then but I don't think you need a new public constructor for this

[witgo]

Oh, do you have a better idea?

[srowen]

Just don't add a new constructor. The existing one can set the new fields.

[witgo]

That will change a lot of code, right?

[srowen]

No, why? I mean keep exactly the same constructors, no more no less. No code would change. You just set your two new fields in the current constructor. It actually means you don't need some of the changes you made here.

[witgo]

I'm sorry, I probably did not make it clear.
Suppose there are E Executor in the cluster, a shuffle process has M Map task, R reduce task, in the master branch will be created:

1. Up to M * R FileSegmentManagedBuffer instances
2. Up to 2 * M * R NoSuchElementException instances

in this PR will be created:

1. Up to M * R FileSegmentManagedBuffer instances
2. Up to 2 * NoSuchElementException instances (ExternalShuffleBlockResolver and IndexShuffleBlockResolver are created once executor starts and They call the new constructor to create a FileSegmentManagedBuffer instance)

[srowen]

This still doesn't address the point. What you say is true even if you make the change I suggest, which is to remove the superfluous constructor. The performance is exactly the same.

[witgo]

Sorry,I didn't get your idea. Can you write some code?

[srowen]

I'm simply describing what you proposed above at #17329 (comment)

[srowen]

Ping @witgo to update or close

[srowen]

OK. I think anyone's welcome to make this change without the new constructor. It sounded fine otherwise.