[SPARK-12511][PySpark][Streaming]Make sure PythonDStream.registerSerializer is called only once

[zsxwing]

There is an issue that Py4J's PythonProxyHandler.finalize blocks forever. (py4j/py4j#184)

Py4j will create a PythonProxyHandler in Java for "transformer_serializer" when calling "registerSerializer". If we call "registerSerializer" twice, the second PythonProxyHandler will override the first one, then the first one will be GCed and trigger "PythonProxyHandler.finalize". To avoid that, we should not call"registerSerializer" more than once, so that "PythonProxyHandler" in Java side won't be GCed.

[zsxwing]

CC @davies

[davies]

LGTM

[SparkQA]

Test build #48429 has finished for PR 10514 at commit 9be0d0a.

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

[zsxwing]

retest this please

[SparkQA]

Test build #48438 has finished for PR 10514 at commit 9be0d0a.

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

[SparkQA]

Test build #48506 has finished for PR 10514 at commit 3c3e164.

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

[zsxwing]

@davies could you take another look? Actually, cls._transformerSerializer's fields need to be updated with the parameters. So my previous approach doesn't work. I updated the PR description for the new approach.

[davies]

Is this targeted for 1.6, or we just wait this to be fixed in py4j? The current change looks over complicated to me.

[zsxwing]

Not sure when py4j will release the next version to fix this one. Looks py4j doesn't release frequently.

[zsxwing]

retest this please

[SparkQA]

Test build #48523 has finished for PR 10514 at commit 340b034.

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

[zsxwing]

I submitted a new commit to fix a similar issue when restarting from checkpoint.

Unlike Scala, PySpark reads the checkpoint twice. Then the PythonProxyHandlers created by the first checkpoint reading will be abandoned and trigger PythonProxyHandler.finalize.

My latest commit eliminated the unnecessary reading to avoid the PythonProxyHandler.finalize issue.

[zsxwing]

Also ping @tdas since you wrote the getOrCreate method.

[SparkQA]

Test build #48512 has finished for PR 10514 at commit 340b034.

• This patch fails from timeout after a configured wait of 250m.
• This patch merges cleanly.
• This patch adds no public classes.

[zsxwing]

retest this please

[SparkQA]

Test build #48544 has finished for PR 10514 at commit 340b034.

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

[davies]

LGTM, do we need to merge this into 1.6?

[zsxwing]

do we need to merge this into 1.6?

Yes since it affects all people using PySpark Streaming checkpoint.

[davies]

Merged into master and 1.6 branch, could you create a JIRA to clean this once the bugs are fixed in py4j (and released)?

[zsxwing]

Merged into master and 1.6 branch, could you create a JIRA to clean this once the bugs are fixed in py4j (and released)?

Created sub tasks in https://issues.apache.org/jira/browse/SPARK-12652