[SPARK-26658][PySpark] : Call pickle.dump with protocol version 3 for Python 3…

[pgandhi999]

… to fix the serialization issue with large objects

When a pyspark job using python 3 tries to serialize large objects, it throws a pickle error in case of trying to serialize global variable object and overflow error in case of broadcast. Refer the corresponding JIRA https://issues.apache.org/jira/browse/SPARK-26658 for more details.

What changes were proposed in this pull request?

Fixed the issue by updating the pickle dump method in code to use protocol version 3 for python 3.

How was this patch tested?

Running manual tests before and after the fix.

Steps To Reproduce:

• To reproduce the above issue, I am using the word2vec model trained on the Google News dataset downloaded from https://drive.google.com/file/d/0B7XkCwpI5KDYNlNUTTlSS21pQmM/edit?usp=sharing

• Use python 3.x with module gensim installed(or ship the module zip file using --py-files).

• Launch pyspark with the following command:
  bin/pyspark --master yarn --py-files additionalPythonModules.zip --conf spark.driver.memory=16g --conf spark.executor.memory=16g --conf spark.driver.memoryOverhead=16g --conf spark.executor.memoryOverhead=16g --conf spark.executor.pyspark.memory=16g

• Run the following commands. For the sake of reproducing the issue, I have simply pasted certain parts of the code here:
  `SparkSession available as 'spark'.

import gensim
score_threshold = 0.65
synonym_limit = 3
model = gensim.models.KeyedVectors.load_word2vec_format('hdfs://home/pgandhi/GoogleNews-vectors-negative300.bin', binary=True)
def isPhrase(word):
... if word.find('_') != -1 :
... return 1
... return 0
...
def process_word(line):
... word = "test"
... positiveWords = []
... positiveWords.append(word)
... try :
... results = model.most_similar(positive=positiveWords)
... synonym_vec = []
... for i in range(len(results)) :
... result = results[i]
... if (result[1] > score_threshold ) :
... synonym = result[0]
... synonym = synonym.lower()
... if (isPhrase(synonym)==0) and (word != synonym) :
... synonym_vec.append(synonym)
... if len(synonym_vec) > synonym_limit :
... break
... if len(synonym_vec) > 0 :
... #print(word +"\t"+ ",".join(synonym_vec))
... return (word, ",".join(synonym_vec))
... except KeyError :
... sys.stderr.write("key error: " + word + "\n")
...
if name == "main":
... rdd = sc.parallelize(["test1", "test2", "test3"])
... rdd2 = rdd.map(process_word)
... rdd2.count()
...`

• For reproducing the issue with broadcast, simply run the code below in pyspark shell:
  `SparkSession available as 'spark'.

import gensim
model = sc.broadcast(gensim.models.KeyedVectors.load_word2vec_format('hdfs://home/pgandhi/GoogleNews-vectors-negative300.bin', binary=True))`

[pgandhi999]

ok to test.

[redsanket]

LGTM, might be nice to copy over how to reproduce section here just a thought so all at one place.

[pgandhi999]

Makes sense @redsanket , thank you.

[redsanket]

guess we can use pickle.DEFAULT_PROTOCOL and pickle.DEFAULT_PROTOCOL - 1 just a thought not sure it matters

[pgandhi999]

The same code exists in serializers.py, I have simply tried to maintain consistency.

[SparkQA]

Test build #101381 has finished for PR 23577 at commit 3204525.

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

[HyukjinKwon]

It duplicates #20691

[HyukjinKwon]

Closing this due to no feedback.