GitHub user BryanCutler opened a pull request:

    [WIP][SPARK-23030][SQL][PYTHON] Use Arrow stream format for creating from 
and collecting Pandas DataFrames

    ## What changes were proposed in this pull request?
    This changes the calls of `toPandas()` and `createDataFrame()` to use the 
Arrow stream format, when Arrow is enabled.  Previously, Arrow data was written 
to byte arrays where each chunk is an output of the Arrow file format.  This 
was mainly due to constraints at the time, and caused some overhead by writing 
the schema/footer on each chunk of data and then having to read multiple Arrow 
file inputs and concat them together.
    Using the Arrow stream format has improved these by increasing performance, 
lower memory overhead for the average case, and simplified the code.  Here are 
the details of this change:
    Spark internal rows are converted to Arrow file format, each group of 
records is a complete Arrow file which contains the schema and other metadata.  
Next a collect is done and an Array of Arrow files is the result.  After that 
each Arrow file is sent to Python driver which then loads each file and concats 
them to a single Arrow DataFrame.
    Spark internal rows are converted to ArrowRecordBatches directly, which is 
the simplest Arrow component for IPC data transfers.  The driver JVM then 
immediately starts serving data to Python as an Arrow stream, sending the 
schema first. It then starts Spark jobs with a custom handler such that when a 
partition is received (and in the correct order) the ArrowRecordBatches can be 
sent to python as soon as possible.  This improves performance, simplifies 
memory usage on executors, and improves the average memory usage on the JVM 
driver.  Since the order of partitions must be preserved, the worst case is 
that the first partition will be the last to arrive and all data must be kept 
in memory until then.  This case is no worse that before when doing a full 
    A Pandas DataFrame is split into parts and each part is made into an Arrow 
file.  Then each file is prefixed by the buffer size and written to a temp 
file.  The temp file is read and each Arrow file is parallelized as a byte 
    A Pandas DataFrame is split into parts, then an Arrow stream is written to 
a temp file where each part is an ArrowRecordBatch.  The temp file is read as a 
stream and the Arrow messages are examined.  If the message is an 
ArrowRecordBatch, the data is saved as a byte array.  After reading the file, 
each ArrowRecordBatch is parallelized as a byte array.  This has slightly more 
processing than before because we must look each Arrow message to extract the 
record batches, but performance remains the same.  It is cleaner in the sense 
that IPC from Python to JVM is done over a single Arrow stream.
    ## How was this patch tested?
    Added new unit tests for the additions to ArrowConverters in Scala, 
existing tests for Python.

You can merge this pull request into a Git repository by running:

    $ git pull 

Alternatively you can review and apply these changes as the patch at:

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #21546
commit 9af482170ee95831bbda139e6e931ba2631df386
Author: Bryan Cutler <cutlerb@...>
Date:   2018-01-10T22:02:15Z

    change ArrowConverters to stream format

commit d617f0da8eff1509da465bb707340e391314bec4
Author: Bryan Cutler <cutlerb@...>
Date:   2018-01-10T22:14:07Z

    Change ArrowSerializer to use stream format

commit f10d5d9cd3cece7f56749e1de7fe01699e4759a0
Author: Bryan Cutler <cutlerb@...>
Date:   2018-01-12T00:40:36Z

    toPandas is working with RecordBatch payloads, using custom handler to 
stream ordered partitions

commit 03653c687473b82bbfb6653504479498a2a3c63b
Author: Bryan Cutler <cutlerb@...>
Date:   2018-02-10T00:23:17Z

    cleanup and removed ArrowPayload, createDataFrame now working

commit 1b932463bca0815e79f3a8d61d1c816e62949698
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-09T00:14:06Z

    toPandas and createDataFrame working but tests fail with date cols

commit ce22d8ad18e052d150528752b727c6cfe11485f7
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-27T00:32:03Z

    removed usage of seekableByteChannel

commit dede0bd96921c439747a9176f24c9ecbb9c8ce0a
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-28T00:28:54Z

    for toPandas, set old collection result to null and add comments

commit 9e29b092cb7d45fa486db0215c3bd4a99c5f8d98
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-28T18:28:18Z

    cleanup, not yet passing ArrowConvertersSuite

commit ceb8d38a6c83c3b6dae040c9e8d860811ecad0cc
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-29T21:14:03Z

    fix to read Arrow stream with multiple batches, cleanup, add docs, scala 
tests pass, style pass

commit f42e4ea7b4fb944eeefd39a0fd6a1428b527214a
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-29T22:17:25Z

    use base OutputStream for serveToStream instead of DataOutputStream

commit 951843d760aa6d29ff18112e82d28f4f6dc09907
Author: Bryan Cutler <cutlerb@...>
Date:   2018-03-29T22:21:13Z

    accidentally removed date type checks, passing pyspark tests

commit af03c6b384fe4ea73d67ad1d3f46be4a1e027e9e
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T00:29:12Z

    Changed to only use Arrow batch bytes as payload, had to hack Arrow 

commit b047c1624429ea579aa279e92909b90400b40c58
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T00:30:48Z

    added todo comment

commit a77b89ea0357e3ce146ff35537eb7da8a8c80bad
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T17:35:05Z

    change getBatchesFromStream to return iterator

commit 81c82093edf78d36a1de850b3d8faede88fb0524
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T18:06:22Z

    need to end stream on toPandas after all batches sent to python, and added 
some comments

commit 5f46a02aa34e9f51ac310d27e3272b883c67cc37
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T18:11:19Z

    forgot to remove old comment

commit 7694b8fe6970789d4e88f9c5df46c97a4b235f02
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T18:47:49Z

    fixed up comments

commit a5a1fbe7121c5b0dd93876a56c29ad17dcd9b168
Author: Bryan Cutler <cutlerb@...>
Date:   2018-06-12T20:22:30Z

    fixed up some wording



To unsubscribe, e-mail:
For additional commands, e-mail:

Reply via email to