support for writing arrow files

[behrica]

This would be super usefull, I believe.
Tech.ml.dataset has very good support for working with arrow files.

(memory mapped and larger then heap)

This would nicely bring Geni and tech.ml.dataset.

I currently can convert with them, using R...

[anthony-khong]

Hi @behrica, I tried to make this happen a couple of times. I think it would be a very good addition to the library. I'm still looking into what the best way of doing it!

[behrica]

Potentially there are 3 ways, I think

1. Extend Spark itself and make it write arrow files.... I searched for this, but did not find anything
2. Make an easy tool for converting the Spark output of Parquet files into arrow files (on file system level)
3. Do "something" in Geni, so the result of "collect" type of calls can be "arrow data structures" (raw bytes or pointers to files or similar) and have tech.ml.dataset ready to create tech.ml.Datasets from this (this is nearly done)

[behrica]

1. and 2) are alternatives. 1) is prefered but maybe imposible to start on.
   Or maybe we need just to wait until the Spark community does it. For now, they seems to be focussing on writing Parquet only

3 could be useful in addition to 1) or 2)

[behrica]

Can 3) potentially work for "data larger the RAM" ?
I don't think so. Spark is not meant to be able to return to the spark driver "large data" (larger then RAM), correct ?
The "work with data larger then RAM capabilities" of Spark are constraint to the "workers", correct ?

[behrica]

I explored 2) and I think it can be done rather easily by using the libraries provided by arrow:
https://github.com/apache/arrow

I have the impression that the C++,Python, R libraries are more advanced then the Java libraries
I looked in detail in to the R package for arrow, and it allows this "transform" in 2 or three lines of code.

But I found "bugs" immediately, but is its foressen and API is there.

[behrica]

Additionally I believe that Spark as optimized data transfers (workers to driver and therefore into a client) into:

• Pandas
• R data frames ([SPARK-25981][R] Enables Arrow optimization from R DataFrame to Spark DataFrame apache/spark#22954)

but not yet for the JVM (Maybe because there is not a single columnar data structure used by everbody)

[behrica]

I was pointed as well to this by Chris Nuerberger:

http://bytedeco.org/javacpp-presets/arrow/apidocs/

This seems to allow to use the C++ API of arrow from java/clojure easely".
This could help for doing 2) and 3)

[anthony-khong]

Can 3) potentially work for "data larger the RAM" ?
I don't think so. Spark is not meant to be able to return to the spark driver "large data" (larger then RAM), correct ?

You're right, I think collect wouldn't work here. And converting formats (i.e. option 2) is also not ideal, because you incur the IO costs twice, once to load and once to write, which is usually non-negligible.

I'm still stumped on this one. Best case scenario is for Spark to support the data format, and we can just call the appropriate methods :)

[behrica]

I saw that Chris started to implement parquet support for t.d.m:
https://github.com/techascent/tech.ml.dataset/tree/parquet-for-real

That is an other option.

[anthony-khong]

Chris is the hero we need, but don't deserve!

[behrica]

I would still think that implementing 2) and accepting that it will not work for larger then RAM is fine.

I still have a lot of "large data" analysis scenarios in my head in which I would use:

• Geni to "filter" the really large data one time into a manageable piece. Maybe even on a "real spark cluster".
  using eventually the Kubernetes way)

• use t.d.m to continue to do "in memory" analysis (eventually off-heap, bu using mmaped arrow files on disk)
  Because t.d.m in-memory "should" still be faster then Spark, potentially at least

(Even Though your benchmark, shows that Geni is slighly faster the "in-memoy t.d.m ...)

So I think to have a function in Geni, which does this:

(-> df
filter
...
...
collect-as-mmaped-arrow-file
)
could be useful.

This should write the spark data frame as an arrow file (or a partition of arrow files) and return a  "file/directory name".

[behrica]

I would still think that implementing 2) and accepting that it will not work for larger then RAM is fine.

I think even larger then RAM can work:
https://www.waitingforcode.com/apache-spark/collecting-part-data-driver-rdd-tolocaiIterator/read_

Not larger then "largest" partition. But that is fine.

[behrica]

So basically it is doing this: https://mikestaszel.com/2018/02/05/writing-huge-csvs-easily-and-efficiently-with-pyspark/
But writing several arrow files, instead of a CSV files.

It will still be slow for very large data, as all data need to flow through the driver. (and crash of too large partitions)

I did not find anything, but it would not surprise me if there is somewhere some java code, which does the RDD to arrow conversion.

[behrica]

Maybe we even can do something else:

The localIterator might allow to:

• convert a rdd row to a clojure map
• use existing t.m.d ->dataset constructor to make a dataset
• write this as nippy file (which seems very fast)

This is of course slow, but as we can re-partition the RDD before in any number of pieces using Spark,
we can control the chunk size

The nippy files can then be opened in TDM and eventually concatinated into a single dataset (if fitting in memory)

[behrica]

I looked into this in detail and I believe that it is indeed doable to write a conversation from:

• the result of (.toLocalIterator fd) (a iteration of type: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
• to something t.m.d can read

I would try to go "to arrow" directly.
It looks rather simple using the Java JDK directly.
https://arrow.apache.org/docs/java/ipc.html#writing-and-reading-streaming-format

The "memory usage" of this is controlled by "2 chunk sizes":

• the "partition size" in Spark: If this is too big, we will get a OOM in the driver at a ceratin point
• the local iterator is un-aware of partitions, is is a continuous stream of rows, if I understand correctly.
  • we needed to decide how many rows to "package" into one arrow file to write
  • this need to be small enough, so that each arrow files fits in RAM
  • I would make this a parameter, to be given by the user (maybe with a default). The ultimate decisions on how big this can be, depends on the knowledge of how size the largest row is....

So we would need a function taking as

• input : seq of org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
• output: seq of org.apache.arrow.vector.ValueVector