[SPARK-10714][SPARK-8632][SPARK-10685][SQL] Refactor Python UDF handling

[rxin]

This patch refactors Python UDF handling:

1. Extract the per-partition Python UDF calling logic from PythonRDD into a PythonRunner. PythonRunner itself expects iterator as input/output, and thus has no dependency on RDD. This way, we can use PythonRunner directly in a mapPartitions call, or in the future in an environment without RDDs.
2. Use PythonRunner in Spark SQL's BatchPythonEvaluation.
3. Updated BatchPythonEvaluation to only use its input once, rather than twice. This should fix Python UDF performance regression in Spark 1.5.

There are a number of small cleanups I wanted to do when I looked at the code, but I kept most of those out so the diff looks small.

This basically implements the approach in #8833, but with some code moving around so the correctness doesn't depend on the inner workings of Spark serialization and task execution.

[rxin]

cc @davies

[SparkQA]

Test build #42712 has finished for PR 8835 at commit 8bfe6c4.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WriterThread(

[davies]

Could you try to re-use the code in Python UDF?

[rxin]

@davies I've updated BatchPythonEvaluation too.

[davies]

I think we should keep the copy() here.

[rxin]

good idea

[SparkQA]

Test build #42719 has finished for PR 8835 at commit 8d3c495.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WriterThread(

[SparkQA]

Test build #42724 has finished for PR 8835 at commit 5e55bf6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class WriterThread(

[JoshRosen]

Could we mitigate this by using a LinkedBlockingDeque to have the producer-side block on inserts once the queue grows to a certain size?

[JoshRosen]

Per discussion offline, the only scenario where the queue can grow really large is when the Python buffer size has been configured to be very large and the UDF result rows are very small. As a result, I think that this comment should be expanded / clarified, but this can take place in a followup PR.

[JoshRosen]

Based on some offline discussion / debate, we've decided to merge this patch into both master and branch-1.5. I'm going to merge this now.