[SPARK-3886] [PySpark] use AutoBatchedSerializer by default

[davies]

Use AutoBatchedSerializer by default, which will choose the proper batch size based on size of serialized objects, let the size of serialized batch fall in into [64k - 640k].

In JVM, the serializer will also track the objects in batch to figure out duplicated objects, larger batch may cause OOM in JVM.

[SparkQA]

QA tests have started for PR 2740 at commit 185f2b9.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2740 at commit 185f2b9.

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

[AmplabJenkins]

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

[JoshRosen]

Spelling: automatically. How about "Set 1 to disable batching, 0 to automatically choose the batch size based on object sizes, or -1 to use an unlimited batch size"?

[JoshRosen]

Aside from a minor doc typo, this looks good to me, especially since AutoBatchedSerializer already exists and has been tested.

[SparkQA]

QA tests have started for PR 2740 at commit 52cdb88.

• This patch merges cleanly.

[JoshRosen]

I tried a small experiment to test this out:

import os
from pyspark import SparkContext, SparkConf

conf = SparkConf().set("spark.executor.memory", "2g")
sc = SparkContext(conf=conf)

mb = 1000000
def inflateDataSize(x):
    return bytearray(os.urandom(1 * mb))

sc.parallelize(range(1000), 10).map(inflateDataSize).cache().count()

Prior to this patch, the Python worker's memory consumption would steadily grow while it attempted to batch together 100 MB of data per task, whereas now the memory usage remains constant because we emit smaller batches more often (since the objects are big).

Thanks for updating the docs. This looks good to me, so I'm going to merge it into master.

[SparkQA]

QA tests have finished for PR 2740 at commit 52cdb88.

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

[AmplabJenkins]

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