...

The first prototype of custom serializers allowed serializers to be chosen on a per-RDD basis. The current implementation only allows one serializer to be used for all data serialization; this serializer is configured when constructing SparkContext.

Wiki MarkupEven with only one serializer, there are still some subtleties here due to how PySpark handles text files. PySpark implements {{SparkContext.textFile()}} by directly calling its Java equivalent. This produces a {{JavaRDD\[String\]}} instead of a {{JavaRDD\[byte\[\]\]}}. JavaRDD transfers these strings to Python workers using Java's MUTF-8 encoding.

{footnote}Prior  <ac:structured-macro ac:name="footnote" ac:schema-version="1" ac:macro-id="6b512aac-84d1-453a-aae4-18bece794ce3"><ac:parameter ac:name="atlassian-macro-output-type">INLINE</ac:parameter><ac:rich-text-body><p>Prior to this pull request, JavaRDD would send strings to Python as pickled UTF-8 strings by prepending the appropriate pickle opcodes.  From the worker's point of view, all of its incoming data was in the same pickled format.  The pull request removed all Python-pickle-specific code from JavaRDD.</p></ac:rich-text-body></ac:structured-macro> To handle these cases, PySpark allows a stage's input deserialization and output serialization functions to come from different serializers.  For example, in {{{footnote}

At the moment, union() requires that its inputs were serialized with the same serializer. When unioning an untransformed RDD created with sc.textFile() with a transformed RDD, cartesian() product, or RDD created with parallelize(), PySpark will force some of the RDDs to be re-serialized using the default serializer. We might be able to add code to avoid this re-serialization, but it would add extra complexity and these particular union() usages seem uncommon.

In the long run, it would be nice to refactor the Java-side serialization logic so that it can apply different interpretations to the bytes that it receives from Python (e.g. unpack them into UTF-8 strings or MsgPack objects). We could also try to remove the assumption that Python sends framed input back to Java, but this this might require a separate socket for sending control messages and exceptions). In the very long term, we might be able to generalize PythonRDD's protocol to a point where we can use the same code to support backends written in other languages (this would effectively be like pipe(), but with a more complicated binary protocol).

{display-footnotes}

Execution and pipelining

PySpark pipelines transformations by composing their functions. When using PySpark, there's a one-to-one correspondence between PySpark stages and Spark scheduler stages. Each PySpark stage corresponds to a PipelinedRDD instance.

...

Why not perform more of the processing in Java?

Wiki MarkupThe first (unpublished) PySpark prototype was implemented in terms of PipedRDD. This prototype represented each Python RDD as a {{JavaRDD\as a JavaRDD[String\]}} of base64-encoded, pickled Python objects. The goal of this approach was to allow functions like join(), distinct(), union(), cogroup(), and groupByKey() to be implemented by directly calling the Java versions.

Wiki MarkupThis approach required some complicated tricks in order to convert the results of Java operations back into pickled data. For example, a leftOuterJoin might produce an {{JavaRDD\[(String, (String, Option\[String\])\]}}:

>>> x = sc.parallelizePairs([("a", 1), ("b", 4)|("a", 1), ("b", 4)])
>>> y = sc.parallelizePairs([("a", 2)|("a", 2)])
>>> print x.leftOuterJoin(y) ._jrdd.collect().toString()
[(UydiJwou,(STQKLg==,None)), (UydhJwou,(STEKLg==,Some(STIKLg==)))|(UydiJwou,(STQKLg==,None)), (UydhJwou,(STEKLg==,Some(STIKLg==)))]

...