You are viewing a plain text version of this content. The canonical link for it is here.

Posted to dev@parquet.apache.org by Matt Massie <ma...@berkeley.edu> on 2015/04/07 20:49:26 UTC

Doing the Spark shuffle on Parquet floors

We are using Apache Parquet and Spark for a genome analysis platform,
called ADAM <http://bdgenomics.org>, that allows researchers to quickly
analyze large datasets of DNA, RNA, etc. Parquet has been a core component
of the system and we see compression of ~20% compared to specialized genome
file formats e.g. compressed BAM. In short, we’re really happy with Parquet.

We are using Avro Specific classes for almost all the entities in our
system, so Avro generates Java classes from our schema
<https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl>.
Since the AvroIndexedRecordConverter has dictionary support, our initial
load from disk to memory is compact.

That’s the good news: compact on-disk and initial in-memory representation.

Here’s the problem: the Spark shuffle.

In order to integrate Parquet with Spark, we use a KryoRegistrator to
register Kryo serializers
<https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala>
for each of our Avro objects (see Kryo Serializer interface
<https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>).
We are serializing each object into record-oriented Avro, which makes our
intermediate shuffle files much larger than the corresponding
column-oriented Parquet inputs. These large shuffle files are hurting our
performance and limiting our scaling for some analysis.

Since the shuffle data is short-lived, there’s no need store meta-data and
we have immediate access to schema through each Avro object. Each Avro
specific class has a SCHEMA$ field which contains the Avro Schema for the
object. There are utility functions in parquet-avro which can convert this
Avro schema into Parquet schema. We also don’t need index pages, only the
dictionary and data pages. We don’t need predicates or projection
functionality. Does anyone on this list see a way to create a Parquet Kryo
serializer
<https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>
to read/write Avro Specific objects to/from a stream? Emitting
column-oriented data will understandably incur memory and CPU costs on the
map side but it will be worth it to improve our shuffle performance.

This shuffle issue to slowing important research so any advice you have to
offer will be appreciated. Thank you.

—
Matt Massie
UC Berkeley, AMPLab

Re: Doing the Spark shuffle on Parquet floors

Posted by Julien Le Dem <ju...@twitter.com.INVALID>.

Matt: I'm looking forward to the blog post that goes with this :)

On Wed, Apr 29, 2015 at 4:36 PM, Dmitriy Ryaboy <dv...@gmail.com> wrote:

> Matt, this is good stuff! :).
>
> On Wed, Apr 29, 2015 at 4:28 PM, Matt Massie <ma...@berkeley.edu> wrote:
>
> > I guess that the correct answer is "it depends" :)
> >
> > Luckily, all the write path costs are being shouldered by the mappers.
> The
> > dictionary encoding will almost certainly decrease the memory/gc pressure
> > we're seeing on the reducers. Without this shuffle manager, Spark
> > serializes all intermediate data into record-oriented blocks and all the
> > Parquet dictionaries are lost. The size of the shuffle data we saw was
> > anywhere from one to two orders of magnitude larger than Parquet inputs
> and
> > the memory used on reducers hurting our performance/scale.
> >
> > Our current anecdotal evidence is that we'll see shuffle sizes about 3x
> the
> > size of inputs (of course, this will vary a lot based on the workload)
> and
> > job completion times that are 3x faster with Parquet in the shuffle.
> >
> > As as aside, since we're using Parquet for input and output files and all
> > intermediate data, we don't use the Spark serializer at all -- it's
> Parquet
> > all the way. :)
> >
> >
> > --
> > Matt Massie
> > UC Berkeley, AMPLab
> >
> > On Wed, Apr 29, 2015 at 3:57 PM, Julien Le Dem
> <julien@twitter.com.invalid
> > >
> > wrote:
> >
> > > This is pretty awesome :)
> > > It's great that the shuffle can be customized in Spark for such cases.
> > > thank you!
> > > I'm curious of the performance. There's some cost to shredding in
> columns
> > > and assembly and savings from the compression (less IO) and dictionary
> > > decoding (less deserialization). I wonder what's the overall change in
> > > time.
> > >
> > > On Wed, Apr 29, 2015 at 3:41 PM, Matt Massie <ma...@berkeley.edu>
> > wrote:
> > >
> > > > I just opened a Jira for a new Spark Parquet shuffle manager.
> > > >
> > > > https://issues.apache.org/jira/browse/SPARK-7263
> > > >
> > > > ...and I have a working end-to-end prototype written at
> > > >
> > > >
> > > >
> > >
> >
> https://github.com/apache/spark/compare/master...massie:parquet-shuffle?expand=1
> > > >
> > > > If anyone is interested in contributing, shoot me an email or comment
> > on
> > > > the Jira.
> > > >
> > > > Thanks again for the great open-source software.
> > > >
> > > >
> > > > --
> > > > Matt Massie
> > > > UC Berkeley, AMPLab
> > > >
> > > > On Thu, Apr 9, 2015 at 11:10 AM, Julien Le Dem
> > > <julien@twitter.com.invalid
> > > > >
> > > > wrote:
> > > >
> > > > > The dictionary is in the first page of a column chunk (if
> dictionary
> > > > > encoded).
> > > > > It would not be very hard to add an API to read it.
> > > > > for a given column and row group:
> > > > >  - seek to the beginning of the column chunk
> > > > >  - read the dictionary page:
> > > > >
> > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
> > > > >  - deserialize the dictionary:
> > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334
> > > > >
> > > > >
> > > > >
> > > > >
> > > > > On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu>
> > > wrote:
> > > > >
> > > > > > Thanks for the ideas, Julien. I appreciate it.
> > > > > >
> > > > > > I suspected that I would need to direct my effort into a
> > > Parquet-based
> > > > > > Spark shuffle but wanted to reach out on the Parquet list first
> to
> > > make
> > > > > > sure that the simpler Parquet serialization approach wouldn't
> work.
> > > > > >
> > > > > > As far as I can tell, there isn't an external API for extracting
> > > > > > dictionaries from Parquet files. Correct?
> > > > > >
> > > > > > -Matt
> > > > > >
> > > > > > On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem
> > > > <julien@twitter.com.invalid
> > > > > >
> > > > > > wrote:
> > > > > >
> > > > > > > You might want to look at Twitter Chill:
> > > > > > https://github.com/twitter/chill
> > > > > > > This is where Twitter has the Kryo serializers for this kind of
> > use
> > > > > > cases.
> > > > > > > The one for Thrift:
> > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > > > > > > Spark is using it already:
> > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> > > > > > >
> > > > > > > But the main problem is that you want to encode multiple
> records
> > > at a
> > > > > > time
> > > > > > > when the serializer gets a single key or value at a time.
> > > > > > > The output is not a single stream since values get routed to a
> > > > > different
> > > > > > > node based on the key.
> > > > > > >
> > > > > > > Unless spark first gets the list of values that goes to a
> > specific
> > > > node
> > > > > > > before encoding them (and provides an API to have custom
> encoding
> > > of
> > > > > all
> > > > > > > values together) I don't think you can implement a Parquet
> > > > > > KryoSerializer.
> > > > > > > I would ask the Spark project whether it's possible to encode
> all
> > > the
> > > > > > > values that go to the same node together with a custom
> > serializer.
> > > > > > >
> > > > > > > Another approach is to reuse dictionary encoding in your row
> > > oriented
> > > > > > > serialization.
> > > > > > > It would not compress as much as Parquet but may help. You'd
> need
> > > to
> > > > > pass
> > > > > > > the dictionary around but maybe you know all the possible
> values
> > > > ahead
> > > > > of
> > > > > > > time. (extracting the dictionaries from the Parquet file?)
> > > > > > > After all your in-memory representation is compact because
> after
> > > > > > dictionary
> > > > > > > decoding all avro struct that use the same string value point
> to
> > > the
> > > > > same
> > > > > > > java string instance, so you may be able to use this.
> > > > > > >
> > > > > > > I hope this helps
> > > > > > >
> > > > > > > Julien
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <
> > massie@berkeley.edu>
> > > > > > wrote:
> > > > > > >
> > > > > > > > I forgot to mention that I’m happy to do the work here, I’m
> > > mostly
> > > > > > > looking
> > > > > > > > for advice and pointers.
> > > > > > > >
> > > > > > > > Also, another interface that might be better for integration
> is
> > > the
> > > > > > Spark
> > > > > > > > Serializer
> > > > > > > > <
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > > > > > > >
> > > > > > > > since the SerializationStream and DeserializationStream have
> > the
> > > > > > flush()
> > > > > > > > and close() functions that are necessary (instead of just
> > > > > read/write).
> > > > > > > > 
> > > > > > > >
> > > > > > > > --
> > > > > > > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > > > > > > <http://www.twitter.com/matt_massie>
> > > > > > > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > > > > > > >
> > > > > > > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <
> > > massie@berkeley.edu>
> > > > > > > wrote:
> > > > > > > >
> > > > > > > > > We are using Apache Parquet and Spark for a genome analysis
> > > > > platform,
> > > > > > > > > called ADAM <http://bdgenomics.org>, that allows
> researchers
> > > to
> > > > > > > quickly
> > > > > > > > > analyze large datasets of DNA, RNA, etc. Parquet has been a
> > > core
> > > > > > > > component
> > > > > > > > > of the system and we see compression of ~20% compared to
> > > > > specialized
> > > > > > > > genome
> > > > > > > > > file formats e.g. compressed BAM. In short, we’re really
> > happy
> > > > with
> > > > > > > > Parquet.
> > > > > > > > >
> > > > > > > > > We are using Avro Specific classes for almost all the
> > entities
> > > in
> > > > > our
> > > > > > > > > system, so Avro generates Java classes from our schema
> > > > > > > > > <
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > > > > > > >.
> > > > > > > > > Since the AvroIndexedRecordConverter has dictionary
> support,
> > > our
> > > > > > > initial
> > > > > > > > > load from disk to memory is compact.
> > > > > > > > >
> > > > > > > > > That’s the good news: compact on-disk and initial in-memory
> > > > > > > > representation.
> > > > > > > > >
> > > > > > > > > Here’s the problem: the Spark shuffle.
> > > > > > > > >
> > > > > > > > > In order to integrate Parquet with Spark, we use a
> > > > KryoRegistrator
> > > > > to
> > > > > > > > > register Kryo serializers
> > > > > > > > > <
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > > > > > > >
> > > > > > > > > for each of our Avro objects (see Kryo Serializer interface
> > > > > > > > > <
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > > > >).
> > > > > > > > > We are serializing each object into record-oriented Avro,
> > which
> > > > > makes
> > > > > > > our
> > > > > > > > > intermediate shuffle files much larger than the
> corresponding
> > > > > > > > > column-oriented Parquet inputs. These large shuffle files
> are
> > > > > hurting
> > > > > > > our
> > > > > > > > > performance and limiting our scaling for some analysis.
> > > > > > > > >
> > > > > > > > > Since the shuffle data is short-lived, there’s no need
> store
> > > > > > meta-data
> > > > > > > > and
> > > > > > > > > we have immediate access to schema through each Avro
> object.
> > > Each
> > > > > > Avro
> > > > > > > > > specific class has a SCHEMA$ field which contains the Avro
> > > Schema
> > > > > for
> > > > > > > the
> > > > > > > > > object. There are utility functions in parquet-avro which
> can
> > > > > convert
> > > > > > > > > this Avro schema into Parquet schema. We also don’t need
> > index
> > > > > pages,
> > > > > > > > only
> > > > > > > > > the dictionary and data pages. We don’t need predicates or
> > > > > projection
> > > > > > > > > functionality. Does anyone on this list see a way to
> create a
> > > > > Parquet
> > > > > > > > Kryo
> > > > > > > > > serializer
> > > > > > > > > <
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > > > >
> > > > > > > > > to read/write Avro Specific objects to/from a stream?
> > Emitting
> > > > > > > > > column-oriented data will understandably incur memory and
> CPU
> > > > costs
> > > > > > on
> > > > > > > > the
> > > > > > > > > map side but it will be worth it to improve our shuffle
> > > > > performance.
> > > > > > > > >
> > > > > > > > > This shuffle issue to slowing important research so any
> > advice
> > > > you
> > > > > > have
> > > > > > > > to
> > > > > > > > > offer will be appreciated. Thank you.
> > > > > > > > >
> > > > > > > > > —
> > > > > > > > > Matt Massie
> > > > > > > > > UC Berkeley, AMPLab
> > > > > > > > > 
> > > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Dmitriy Ryaboy <dv...@gmail.com>.

Matt, this is good stuff! :).

On Wed, Apr 29, 2015 at 4:28 PM, Matt Massie <ma...@berkeley.edu> wrote:

> I guess that the correct answer is "it depends" :)
>
> Luckily, all the write path costs are being shouldered by the mappers. The
> dictionary encoding will almost certainly decrease the memory/gc pressure
> we're seeing on the reducers. Without this shuffle manager, Spark
> serializes all intermediate data into record-oriented blocks and all the
> Parquet dictionaries are lost. The size of the shuffle data we saw was
> anywhere from one to two orders of magnitude larger than Parquet inputs and
> the memory used on reducers hurting our performance/scale.
>
> Our current anecdotal evidence is that we'll see shuffle sizes about 3x the
> size of inputs (of course, this will vary a lot based on the workload) and
> job completion times that are 3x faster with Parquet in the shuffle.
>
> As as aside, since we're using Parquet for input and output files and all
> intermediate data, we don't use the Spark serializer at all -- it's Parquet
> all the way. :)
>
>
> --
> Matt Massie
> UC Berkeley, AMPLab
>
> On Wed, Apr 29, 2015 at 3:57 PM, Julien Le Dem <julien@twitter.com.invalid
> >
> wrote:
>
> > This is pretty awesome :)
> > It's great that the shuffle can be customized in Spark for such cases.
> > thank you!
> > I'm curious of the performance. There's some cost to shredding in columns
> > and assembly and savings from the compression (less IO) and dictionary
> > decoding (less deserialization). I wonder what's the overall change in
> > time.
> >
> > On Wed, Apr 29, 2015 at 3:41 PM, Matt Massie <ma...@berkeley.edu>
> wrote:
> >
> > > I just opened a Jira for a new Spark Parquet shuffle manager.
> > >
> > > https://issues.apache.org/jira/browse/SPARK-7263
> > >
> > > ...and I have a working end-to-end prototype written at
> > >
> > >
> > >
> >
> https://github.com/apache/spark/compare/master...massie:parquet-shuffle?expand=1
> > >
> > > If anyone is interested in contributing, shoot me an email or comment
> on
> > > the Jira.
> > >
> > > Thanks again for the great open-source software.
> > >
> > >
> > > --
> > > Matt Massie
> > > UC Berkeley, AMPLab
> > >
> > > On Thu, Apr 9, 2015 at 11:10 AM, Julien Le Dem
> > <julien@twitter.com.invalid
> > > >
> > > wrote:
> > >
> > > > The dictionary is in the first page of a column chunk (if dictionary
> > > > encoded).
> > > > It would not be very hard to add an API to read it.
> > > > for a given column and row group:
> > > >  - seek to the beginning of the column chunk
> > > >  - read the dictionary page:
> > > >
> > > >
> > > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
> > > >  - deserialize the dictionary:
> > > >
> > > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334
> > > >
> > > >
> > > >
> > > >
> > > > On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu>
> > wrote:
> > > >
> > > > > Thanks for the ideas, Julien. I appreciate it.
> > > > >
> > > > > I suspected that I would need to direct my effort into a
> > Parquet-based
> > > > > Spark shuffle but wanted to reach out on the Parquet list first to
> > make
> > > > > sure that the simpler Parquet serialization approach wouldn't work.
> > > > >
> > > > > As far as I can tell, there isn't an external API for extracting
> > > > > dictionaries from Parquet files. Correct?
> > > > >
> > > > > -Matt
> > > > >
> > > > > On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem
> > > <julien@twitter.com.invalid
> > > > >
> > > > > wrote:
> > > > >
> > > > > > You might want to look at Twitter Chill:
> > > > > https://github.com/twitter/chill
> > > > > > This is where Twitter has the Kryo serializers for this kind of
> use
> > > > > cases.
> > > > > > The one for Thrift:
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > > > > > Spark is using it already:
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> > > > > >
> > > > > > But the main problem is that you want to encode multiple records
> > at a
> > > > > time
> > > > > > when the serializer gets a single key or value at a time.
> > > > > > The output is not a single stream since values get routed to a
> > > > different
> > > > > > node based on the key.
> > > > > >
> > > > > > Unless spark first gets the list of values that goes to a
> specific
> > > node
> > > > > > before encoding them (and provides an API to have custom encoding
> > of
> > > > all
> > > > > > values together) I don't think you can implement a Parquet
> > > > > KryoSerializer.
> > > > > > I would ask the Spark project whether it's possible to encode all
> > the
> > > > > > values that go to the same node together with a custom
> serializer.
> > > > > >
> > > > > > Another approach is to reuse dictionary encoding in your row
> > oriented
> > > > > > serialization.
> > > > > > It would not compress as much as Parquet but may help. You'd need
> > to
> > > > pass
> > > > > > the dictionary around but maybe you know all the possible values
> > > ahead
> > > > of
> > > > > > time. (extracting the dictionaries from the Parquet file?)
> > > > > > After all your in-memory representation is compact because after
> > > > > dictionary
> > > > > > decoding all avro struct that use the same string value point to
> > the
> > > > same
> > > > > > java string instance, so you may be able to use this.
> > > > > >
> > > > > > I hope this helps
> > > > > >
> > > > > > Julien
> > > > > >
> > > > > >
> > > > > >
> > > > > > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <
> massie@berkeley.edu>
> > > > > wrote:
> > > > > >
> > > > > > > I forgot to mention that I’m happy to do the work here, I’m
> > mostly
> > > > > > looking
> > > > > > > for advice and pointers.
> > > > > > >
> > > > > > > Also, another interface that might be better for integration is
> > the
> > > > > Spark
> > > > > > > Serializer
> > > > > > > <
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > > > > > >
> > > > > > > since the SerializationStream and DeserializationStream have
> the
> > > > > flush()
> > > > > > > and close() functions that are necessary (instead of just
> > > > read/write).
> > > > > > > 
> > > > > > >
> > > > > > > --
> > > > > > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > > > > > <http://www.twitter.com/matt_massie>
> > > > > > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > > > > > >
> > > > > > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <
> > massie@berkeley.edu>
> > > > > > wrote:
> > > > > > >
> > > > > > > > We are using Apache Parquet and Spark for a genome analysis
> > > > platform,
> > > > > > > > called ADAM <http://bdgenomics.org>, that allows researchers
> > to
> > > > > > quickly
> > > > > > > > analyze large datasets of DNA, RNA, etc. Parquet has been a
> > core
> > > > > > > component
> > > > > > > > of the system and we see compression of ~20% compared to
> > > > specialized
> > > > > > > genome
> > > > > > > > file formats e.g. compressed BAM. In short, we’re really
> happy
> > > with
> > > > > > > Parquet.
> > > > > > > >
> > > > > > > > We are using Avro Specific classes for almost all the
> entities
> > in
> > > > our
> > > > > > > > system, so Avro generates Java classes from our schema
> > > > > > > > <
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > > > > > >.
> > > > > > > > Since the AvroIndexedRecordConverter has dictionary support,
> > our
> > > > > > initial
> > > > > > > > load from disk to memory is compact.
> > > > > > > >
> > > > > > > > That’s the good news: compact on-disk and initial in-memory
> > > > > > > representation.
> > > > > > > >
> > > > > > > > Here’s the problem: the Spark shuffle.
> > > > > > > >
> > > > > > > > In order to integrate Parquet with Spark, we use a
> > > KryoRegistrator
> > > > to
> > > > > > > > register Kryo serializers
> > > > > > > > <
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > > > > > >
> > > > > > > > for each of our Avro objects (see Kryo Serializer interface
> > > > > > > > <
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > > >).
> > > > > > > > We are serializing each object into record-oriented Avro,
> which
> > > > makes
> > > > > > our
> > > > > > > > intermediate shuffle files much larger than the corresponding
> > > > > > > > column-oriented Parquet inputs. These large shuffle files are
> > > > hurting
> > > > > > our
> > > > > > > > performance and limiting our scaling for some analysis.
> > > > > > > >
> > > > > > > > Since the shuffle data is short-lived, there’s no need store
> > > > > meta-data
> > > > > > > and
> > > > > > > > we have immediate access to schema through each Avro object.
> > Each
> > > > > Avro
> > > > > > > > specific class has a SCHEMA$ field which contains the Avro
> > Schema
> > > > for
> > > > > > the
> > > > > > > > object. There are utility functions in parquet-avro which can
> > > > convert
> > > > > > > > this Avro schema into Parquet schema. We also don’t need
> index
> > > > pages,
> > > > > > > only
> > > > > > > > the dictionary and data pages. We don’t need predicates or
> > > > projection
> > > > > > > > functionality. Does anyone on this list see a way to create a
> > > > Parquet
> > > > > > > Kryo
> > > > > > > > serializer
> > > > > > > > <
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > > >
> > > > > > > > to read/write Avro Specific objects to/from a stream?
> Emitting
> > > > > > > > column-oriented data will understandably incur memory and CPU
> > > costs
> > > > > on
> > > > > > > the
> > > > > > > > map side but it will be worth it to improve our shuffle
> > > > performance.
> > > > > > > >
> > > > > > > > This shuffle issue to slowing important research so any
> advice
> > > you
> > > > > have
> > > > > > > to
> > > > > > > > offer will be appreciated. Thank you.
> > > > > > > >
> > > > > > > > —
> > > > > > > > Matt Massie
> > > > > > > > UC Berkeley, AMPLab
> > > > > > > > 
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Matt Massie <ma...@berkeley.edu>.

I guess that the correct answer is "it depends" :)

Luckily, all the write path costs are being shouldered by the mappers. The
dictionary encoding will almost certainly decrease the memory/gc pressure
we're seeing on the reducers. Without this shuffle manager, Spark
serializes all intermediate data into record-oriented blocks and all the
Parquet dictionaries are lost. The size of the shuffle data we saw was
anywhere from one to two orders of magnitude larger than Parquet inputs and
the memory used on reducers hurting our performance/scale.

Our current anecdotal evidence is that we'll see shuffle sizes about 3x the
size of inputs (of course, this will vary a lot based on the workload) and
job completion times that are 3x faster with Parquet in the shuffle.

As as aside, since we're using Parquet for input and output files and all
intermediate data, we don't use the Spark serializer at all -- it's Parquet
all the way. :)


--
Matt Massie
UC Berkeley, AMPLab

On Wed, Apr 29, 2015 at 3:57 PM, Julien Le Dem <ju...@twitter.com.invalid>
wrote:

> This is pretty awesome :)
> It's great that the shuffle can be customized in Spark for such cases.
> thank you!
> I'm curious of the performance. There's some cost to shredding in columns
> and assembly and savings from the compression (less IO) and dictionary
> decoding (less deserialization). I wonder what's the overall change in
> time.
>
> On Wed, Apr 29, 2015 at 3:41 PM, Matt Massie <ma...@berkeley.edu> wrote:
>
> > I just opened a Jira for a new Spark Parquet shuffle manager.
> >
> > https://issues.apache.org/jira/browse/SPARK-7263
> >
> > ...and I have a working end-to-end prototype written at
> >
> >
> >
> https://github.com/apache/spark/compare/master...massie:parquet-shuffle?expand=1
> >
> > If anyone is interested in contributing, shoot me an email or comment on
> > the Jira.
> >
> > Thanks again for the great open-source software.
> >
> >
> > --
> > Matt Massie
> > UC Berkeley, AMPLab
> >
> > On Thu, Apr 9, 2015 at 11:10 AM, Julien Le Dem
> <julien@twitter.com.invalid
> > >
> > wrote:
> >
> > > The dictionary is in the first page of a column chunk (if dictionary
> > > encoded).
> > > It would not be very hard to add an API to read it.
> > > for a given column and row group:
> > >  - seek to the beginning of the column chunk
> > >  - read the dictionary page:
> > >
> > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
> > >  - deserialize the dictionary:
> > >
> > >
> >
> https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334
> > >
> > >
> > >
> > >
> > > On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu>
> wrote:
> > >
> > > > Thanks for the ideas, Julien. I appreciate it.
> > > >
> > > > I suspected that I would need to direct my effort into a
> Parquet-based
> > > > Spark shuffle but wanted to reach out on the Parquet list first to
> make
> > > > sure that the simpler Parquet serialization approach wouldn't work.
> > > >
> > > > As far as I can tell, there isn't an external API for extracting
> > > > dictionaries from Parquet files. Correct?
> > > >
> > > > -Matt
> > > >
> > > > On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem
> > <julien@twitter.com.invalid
> > > >
> > > > wrote:
> > > >
> > > > > You might want to look at Twitter Chill:
> > > > https://github.com/twitter/chill
> > > > > This is where Twitter has the Kryo serializers for this kind of use
> > > > cases.
> > > > > The one for Thrift:
> > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > > > > Spark is using it already:
> > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> > > > >
> > > > > But the main problem is that you want to encode multiple records
> at a
> > > > time
> > > > > when the serializer gets a single key or value at a time.
> > > > > The output is not a single stream since values get routed to a
> > > different
> > > > > node based on the key.
> > > > >
> > > > > Unless spark first gets the list of values that goes to a specific
> > node
> > > > > before encoding them (and provides an API to have custom encoding
> of
> > > all
> > > > > values together) I don't think you can implement a Parquet
> > > > KryoSerializer.
> > > > > I would ask the Spark project whether it's possible to encode all
> the
> > > > > values that go to the same node together with a custom serializer.
> > > > >
> > > > > Another approach is to reuse dictionary encoding in your row
> oriented
> > > > > serialization.
> > > > > It would not compress as much as Parquet but may help. You'd need
> to
> > > pass
> > > > > the dictionary around but maybe you know all the possible values
> > ahead
> > > of
> > > > > time. (extracting the dictionaries from the Parquet file?)
> > > > > After all your in-memory representation is compact because after
> > > > dictionary
> > > > > decoding all avro struct that use the same string value point to
> the
> > > same
> > > > > java string instance, so you may be able to use this.
> > > > >
> > > > > I hope this helps
> > > > >
> > > > > Julien
> > > > >
> > > > >
> > > > >
> > > > > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu>
> > > > wrote:
> > > > >
> > > > > > I forgot to mention that I’m happy to do the work here, I’m
> mostly
> > > > > looking
> > > > > > for advice and pointers.
> > > > > >
> > > > > > Also, another interface that might be better for integration is
> the
> > > > Spark
> > > > > > Serializer
> > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > > > > >
> > > > > > since the SerializationStream and DeserializationStream have the
> > > > flush()
> > > > > > and close() functions that are necessary (instead of just
> > > read/write).
> > > > > > 
> > > > > >
> > > > > > --
> > > > > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > > > > <http://www.twitter.com/matt_massie>
> > > > > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > > > > >
> > > > > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <
> massie@berkeley.edu>
> > > > > wrote:
> > > > > >
> > > > > > > We are using Apache Parquet and Spark for a genome analysis
> > > platform,
> > > > > > > called ADAM <http://bdgenomics.org>, that allows researchers
> to
> > > > > quickly
> > > > > > > analyze large datasets of DNA, RNA, etc. Parquet has been a
> core
> > > > > > component
> > > > > > > of the system and we see compression of ~20% compared to
> > > specialized
> > > > > > genome
> > > > > > > file formats e.g. compressed BAM. In short, we’re really happy
> > with
> > > > > > Parquet.
> > > > > > >
> > > > > > > We are using Avro Specific classes for almost all the entities
> in
> > > our
> > > > > > > system, so Avro generates Java classes from our schema
> > > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > > > > >.
> > > > > > > Since the AvroIndexedRecordConverter has dictionary support,
> our
> > > > > initial
> > > > > > > load from disk to memory is compact.
> > > > > > >
> > > > > > > That’s the good news: compact on-disk and initial in-memory
> > > > > > representation.
> > > > > > >
> > > > > > > Here’s the problem: the Spark shuffle.
> > > > > > >
> > > > > > > In order to integrate Parquet with Spark, we use a
> > KryoRegistrator
> > > to
> > > > > > > register Kryo serializers
> > > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > > > > >
> > > > > > > for each of our Avro objects (see Kryo Serializer interface
> > > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > >).
> > > > > > > We are serializing each object into record-oriented Avro, which
> > > makes
> > > > > our
> > > > > > > intermediate shuffle files much larger than the corresponding
> > > > > > > column-oriented Parquet inputs. These large shuffle files are
> > > hurting
> > > > > our
> > > > > > > performance and limiting our scaling for some analysis.
> > > > > > >
> > > > > > > Since the shuffle data is short-lived, there’s no need store
> > > > meta-data
> > > > > > and
> > > > > > > we have immediate access to schema through each Avro object.
> Each
> > > > Avro
> > > > > > > specific class has a SCHEMA$ field which contains the Avro
> Schema
> > > for
> > > > > the
> > > > > > > object. There are utility functions in parquet-avro which can
> > > convert
> > > > > > > this Avro schema into Parquet schema. We also don’t need index
> > > pages,
> > > > > > only
> > > > > > > the dictionary and data pages. We don’t need predicates or
> > > projection
> > > > > > > functionality. Does anyone on this list see a way to create a
> > > Parquet
> > > > > > Kryo
> > > > > > > serializer
> > > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > > >
> > > > > > > to read/write Avro Specific objects to/from a stream? Emitting
> > > > > > > column-oriented data will understandably incur memory and CPU
> > costs
> > > > on
> > > > > > the
> > > > > > > map side but it will be worth it to improve our shuffle
> > > performance.
> > > > > > >
> > > > > > > This shuffle issue to slowing important research so any advice
> > you
> > > > have
> > > > > > to
> > > > > > > offer will be appreciated. Thank you.
> > > > > > >
> > > > > > > —
> > > > > > > Matt Massie
> > > > > > > UC Berkeley, AMPLab
> > > > > > > 
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Julien Le Dem <ju...@twitter.com.INVALID>.

This is pretty awesome :)
It's great that the shuffle can be customized in Spark for such cases.
thank you!
I'm curious of the performance. There's some cost to shredding in columns
and assembly and savings from the compression (less IO) and dictionary
decoding (less deserialization). I wonder what's the overall change in time.

On Wed, Apr 29, 2015 at 3:41 PM, Matt Massie <ma...@berkeley.edu> wrote:

> I just opened a Jira for a new Spark Parquet shuffle manager.
>
> https://issues.apache.org/jira/browse/SPARK-7263
>
> ...and I have a working end-to-end prototype written at
>
>
> https://github.com/apache/spark/compare/master...massie:parquet-shuffle?expand=1
>
> If anyone is interested in contributing, shoot me an email or comment on
> the Jira.
>
> Thanks again for the great open-source software.
>
>
> --
> Matt Massie
> UC Berkeley, AMPLab
>
> On Thu, Apr 9, 2015 at 11:10 AM, Julien Le Dem <julien@twitter.com.invalid
> >
> wrote:
>
> > The dictionary is in the first page of a column chunk (if dictionary
> > encoded).
> > It would not be very hard to add an API to read it.
> > for a given column and row group:
> >  - seek to the beginning of the column chunk
> >  - read the dictionary page:
> >
> >
> >
> https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
> >  - deserialize the dictionary:
> >
> >
> https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334
> >
> >
> >
> >
> > On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu> wrote:
> >
> > > Thanks for the ideas, Julien. I appreciate it.
> > >
> > > I suspected that I would need to direct my effort into a Parquet-based
> > > Spark shuffle but wanted to reach out on the Parquet list first to make
> > > sure that the simpler Parquet serialization approach wouldn't work.
> > >
> > > As far as I can tell, there isn't an external API for extracting
> > > dictionaries from Parquet files. Correct?
> > >
> > > -Matt
> > >
> > > On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem
> <julien@twitter.com.invalid
> > >
> > > wrote:
> > >
> > > > You might want to look at Twitter Chill:
> > > https://github.com/twitter/chill
> > > > This is where Twitter has the Kryo serializers for this kind of use
> > > cases.
> > > > The one for Thrift:
> > > >
> > > >
> > >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > > > Spark is using it already:
> > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> > > >
> > > > But the main problem is that you want to encode multiple records at a
> > > time
> > > > when the serializer gets a single key or value at a time.
> > > > The output is not a single stream since values get routed to a
> > different
> > > > node based on the key.
> > > >
> > > > Unless spark first gets the list of values that goes to a specific
> node
> > > > before encoding them (and provides an API to have custom encoding of
> > all
> > > > values together) I don't think you can implement a Parquet
> > > KryoSerializer.
> > > > I would ask the Spark project whether it's possible to encode all the
> > > > values that go to the same node together with a custom serializer.
> > > >
> > > > Another approach is to reuse dictionary encoding in your row oriented
> > > > serialization.
> > > > It would not compress as much as Parquet but may help. You'd need to
> > pass
> > > > the dictionary around but maybe you know all the possible values
> ahead
> > of
> > > > time. (extracting the dictionaries from the Parquet file?)
> > > > After all your in-memory representation is compact because after
> > > dictionary
> > > > decoding all avro struct that use the same string value point to the
> > same
> > > > java string instance, so you may be able to use this.
> > > >
> > > > I hope this helps
> > > >
> > > > Julien
> > > >
> > > >
> > > >
> > > > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu>
> > > wrote:
> > > >
> > > > > I forgot to mention that I’m happy to do the work here, I’m mostly
> > > > looking
> > > > > for advice and pointers.
> > > > >
> > > > > Also, another interface that might be better for integration is the
> > > Spark
> > > > > Serializer
> > > > > <
> > > > >
> > > >
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > > > >
> > > > > since the SerializationStream and DeserializationStream have the
> > > flush()
> > > > > and close() functions that are necessary (instead of just
> > read/write).
> > > > > 
> > > > >
> > > > > --
> > > > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > > > <http://www.twitter.com/matt_massie>
> > > > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > > > >
> > > > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu>
> > > > wrote:
> > > > >
> > > > > > We are using Apache Parquet and Spark for a genome analysis
> > platform,
> > > > > > called ADAM <http://bdgenomics.org>, that allows researchers to
> > > > quickly
> > > > > > analyze large datasets of DNA, RNA, etc. Parquet has been a core
> > > > > component
> > > > > > of the system and we see compression of ~20% compared to
> > specialized
> > > > > genome
> > > > > > file formats e.g. compressed BAM. In short, we’re really happy
> with
> > > > > Parquet.
> > > > > >
> > > > > > We are using Avro Specific classes for almost all the entities in
> > our
> > > > > > system, so Avro generates Java classes from our schema
> > > > > > <
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > > > >.
> > > > > > Since the AvroIndexedRecordConverter has dictionary support, our
> > > > initial
> > > > > > load from disk to memory is compact.
> > > > > >
> > > > > > That’s the good news: compact on-disk and initial in-memory
> > > > > representation.
> > > > > >
> > > > > > Here’s the problem: the Spark shuffle.
> > > > > >
> > > > > > In order to integrate Parquet with Spark, we use a
> KryoRegistrator
> > to
> > > > > > register Kryo serializers
> > > > > > <
> > > > >
> > > >
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > > > >
> > > > > > for each of our Avro objects (see Kryo Serializer interface
> > > > > > <
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > >).
> > > > > > We are serializing each object into record-oriented Avro, which
> > makes
> > > > our
> > > > > > intermediate shuffle files much larger than the corresponding
> > > > > > column-oriented Parquet inputs. These large shuffle files are
> > hurting
> > > > our
> > > > > > performance and limiting our scaling for some analysis.
> > > > > >
> > > > > > Since the shuffle data is short-lived, there’s no need store
> > > meta-data
> > > > > and
> > > > > > we have immediate access to schema through each Avro object. Each
> > > Avro
> > > > > > specific class has a SCHEMA$ field which contains the Avro Schema
> > for
> > > > the
> > > > > > object. There are utility functions in parquet-avro which can
> > convert
> > > > > > this Avro schema into Parquet schema. We also don’t need index
> > pages,
> > > > > only
> > > > > > the dictionary and data pages. We don’t need predicates or
> > projection
> > > > > > functionality. Does anyone on this list see a way to create a
> > Parquet
> > > > > Kryo
> > > > > > serializer
> > > > > > <
> > > > >
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > > >
> > > > > > to read/write Avro Specific objects to/from a stream? Emitting
> > > > > > column-oriented data will understandably incur memory and CPU
> costs
> > > on
> > > > > the
> > > > > > map side but it will be worth it to improve our shuffle
> > performance.
> > > > > >
> > > > > > This shuffle issue to slowing important research so any advice
> you
> > > have
> > > > > to
> > > > > > offer will be appreciated. Thank you.
> > > > > >
> > > > > > —
> > > > > > Matt Massie
> > > > > > UC Berkeley, AMPLab
> > > > > > 
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Matt Massie <ma...@berkeley.edu>.

I just opened a Jira for a new Spark Parquet shuffle manager.

https://issues.apache.org/jira/browse/SPARK-7263

...and I have a working end-to-end prototype written at

https://github.com/apache/spark/compare/master...massie:parquet-shuffle?expand=1

If anyone is interested in contributing, shoot me an email or comment on
the Jira.

Thanks again for the great open-source software.


--
Matt Massie
UC Berkeley, AMPLab

On Thu, Apr 9, 2015 at 11:10 AM, Julien Le Dem <ju...@twitter.com.invalid>
wrote:

> The dictionary is in the first page of a column chunk (if dictionary
> encoded).
> It would not be very hard to add an API to read it.
> for a given column and row group:
>  - seek to the beginning of the column chunk
>  - read the dictionary page:
>
>
> https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
>  - deserialize the dictionary:
>
> https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334
>
>
>
>
> On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu> wrote:
>
> > Thanks for the ideas, Julien. I appreciate it.
> >
> > I suspected that I would need to direct my effort into a Parquet-based
> > Spark shuffle but wanted to reach out on the Parquet list first to make
> > sure that the simpler Parquet serialization approach wouldn't work.
> >
> > As far as I can tell, there isn't an external API for extracting
> > dictionaries from Parquet files. Correct?
> >
> > -Matt
> >
> > On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem <julien@twitter.com.invalid
> >
> > wrote:
> >
> > > You might want to look at Twitter Chill:
> > https://github.com/twitter/chill
> > > This is where Twitter has the Kryo serializers for this kind of use
> > cases.
> > > The one for Thrift:
> > >
> > >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > > Spark is using it already:
> > >
> > >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> > >
> > > But the main problem is that you want to encode multiple records at a
> > time
> > > when the serializer gets a single key or value at a time.
> > > The output is not a single stream since values get routed to a
> different
> > > node based on the key.
> > >
> > > Unless spark first gets the list of values that goes to a specific node
> > > before encoding them (and provides an API to have custom encoding of
> all
> > > values together) I don't think you can implement a Parquet
> > KryoSerializer.
> > > I would ask the Spark project whether it's possible to encode all the
> > > values that go to the same node together with a custom serializer.
> > >
> > > Another approach is to reuse dictionary encoding in your row oriented
> > > serialization.
> > > It would not compress as much as Parquet but may help. You'd need to
> pass
> > > the dictionary around but maybe you know all the possible values ahead
> of
> > > time. (extracting the dictionaries from the Parquet file?)
> > > After all your in-memory representation is compact because after
> > dictionary
> > > decoding all avro struct that use the same string value point to the
> same
> > > java string instance, so you may be able to use this.
> > >
> > > I hope this helps
> > >
> > > Julien
> > >
> > >
> > >
> > > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu>
> > wrote:
> > >
> > > > I forgot to mention that I’m happy to do the work here, I’m mostly
> > > looking
> > > > for advice and pointers.
> > > >
> > > > Also, another interface that might be better for integration is the
> > Spark
> > > > Serializer
> > > > <
> > > >
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > > >
> > > > since the SerializationStream and DeserializationStream have the
> > flush()
> > > > and close() functions that are necessary (instead of just
> read/write).
> > > > 
> > > >
> > > > --
> > > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > > <http://www.twitter.com/matt_massie>
> > > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > > >
> > > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu>
> > > wrote:
> > > >
> > > > > We are using Apache Parquet and Spark for a genome analysis
> platform,
> > > > > called ADAM <http://bdgenomics.org>, that allows researchers to
> > > quickly
> > > > > analyze large datasets of DNA, RNA, etc. Parquet has been a core
> > > > component
> > > > > of the system and we see compression of ~20% compared to
> specialized
> > > > genome
> > > > > file formats e.g. compressed BAM. In short, we’re really happy with
> > > > Parquet.
> > > > >
> > > > > We are using Avro Specific classes for almost all the entities in
> our
> > > > > system, so Avro generates Java classes from our schema
> > > > > <
> > > >
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > > >.
> > > > > Since the AvroIndexedRecordConverter has dictionary support, our
> > > initial
> > > > > load from disk to memory is compact.
> > > > >
> > > > > That’s the good news: compact on-disk and initial in-memory
> > > > representation.
> > > > >
> > > > > Here’s the problem: the Spark shuffle.
> > > > >
> > > > > In order to integrate Parquet with Spark, we use a KryoRegistrator
> to
> > > > > register Kryo serializers
> > > > > <
> > > >
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > > >
> > > > > for each of our Avro objects (see Kryo Serializer interface
> > > > > <
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > >).
> > > > > We are serializing each object into record-oriented Avro, which
> makes
> > > our
> > > > > intermediate shuffle files much larger than the corresponding
> > > > > column-oriented Parquet inputs. These large shuffle files are
> hurting
> > > our
> > > > > performance and limiting our scaling for some analysis.
> > > > >
> > > > > Since the shuffle data is short-lived, there’s no need store
> > meta-data
> > > > and
> > > > > we have immediate access to schema through each Avro object. Each
> > Avro
> > > > > specific class has a SCHEMA$ field which contains the Avro Schema
> for
> > > the
> > > > > object. There are utility functions in parquet-avro which can
> convert
> > > > > this Avro schema into Parquet schema. We also don’t need index
> pages,
> > > > only
> > > > > the dictionary and data pages. We don’t need predicates or
> projection
> > > > > functionality. Does anyone on this list see a way to create a
> Parquet
> > > > Kryo
> > > > > serializer
> > > > > <
> > > >
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > > >
> > > > > to read/write Avro Specific objects to/from a stream? Emitting
> > > > > column-oriented data will understandably incur memory and CPU costs
> > on
> > > > the
> > > > > map side but it will be worth it to improve our shuffle
> performance.
> > > > >
> > > > > This shuffle issue to slowing important research so any advice you
> > have
> > > > to
> > > > > offer will be appreciated. Thank you.
> > > > >
> > > > > —
> > > > > Matt Massie
> > > > > UC Berkeley, AMPLab
> > > > > 
> > > > >
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Julien Le Dem <ju...@twitter.com.INVALID>.

The dictionary is in the first page of a column chunk (if dictionary
encoded).
It would not be very hard to add an API to read it.
for a given column and row group:
 - seek to the beginning of the column chunk
 - read the dictionary page:

https://github.com/apache/incubator-parquet-mr/blob/master/parquet-hadoop/src/main/java/parquet/hadoop/ColumnChunkPageReadStore.java
 - deserialize the dictionary:
https://github.com/apache/incubator-parquet-mr/blob/3df3372a1ee7b6ea74af89f53a614895b8078609/parquet-column/src/main/java/parquet/column/impl/ColumnReaderImpl.java#L334




On Wed, Apr 8, 2015 at 1:28 PM, Matt Massie <ma...@berkeley.edu> wrote:

> Thanks for the ideas, Julien. I appreciate it.
>
> I suspected that I would need to direct my effort into a Parquet-based
> Spark shuffle but wanted to reach out on the Parquet list first to make
> sure that the simpler Parquet serialization approach wouldn't work.
>
> As far as I can tell, there isn't an external API for extracting
> dictionaries from Parquet files. Correct?
>
> -Matt
>
> On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem <ju...@twitter.com.invalid>
> wrote:
>
> > You might want to look at Twitter Chill:
> https://github.com/twitter/chill
> > This is where Twitter has the Kryo serializers for this kind of use
> cases.
> > The one for Thrift:
> >
> >
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> > Spark is using it already:
> >
> >
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
> >
> > But the main problem is that you want to encode multiple records at a
> time
> > when the serializer gets a single key or value at a time.
> > The output is not a single stream since values get routed to a different
> > node based on the key.
> >
> > Unless spark first gets the list of values that goes to a specific node
> > before encoding them (and provides an API to have custom encoding of all
> > values together) I don't think you can implement a Parquet
> KryoSerializer.
> > I would ask the Spark project whether it's possible to encode all the
> > values that go to the same node together with a custom serializer.
> >
> > Another approach is to reuse dictionary encoding in your row oriented
> > serialization.
> > It would not compress as much as Parquet but may help. You'd need to pass
> > the dictionary around but maybe you know all the possible values ahead of
> > time. (extracting the dictionaries from the Parquet file?)
> > After all your in-memory representation is compact because after
> dictionary
> > decoding all avro struct that use the same string value point to the same
> > java string instance, so you may be able to use this.
> >
> > I hope this helps
> >
> > Julien
> >
> >
> >
> > On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu>
> wrote:
> >
> > > I forgot to mention that I’m happy to do the work here, I’m mostly
> > looking
> > > for advice and pointers.
> > >
> > > Also, another interface that might be better for integration is the
> Spark
> > > Serializer
> > > <
> > >
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > > >
> > > since the SerializationStream and DeserializationStream have the
> flush()
> > > and close() functions that are necessary (instead of just read/write).
> > > 
> > >
> > > --
> > > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > > <http://www.twitter.com/matt_massie>
> > > UC, Berkeley AMPLab <https://twitter.com/amplab>
> > >
> > > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu>
> > wrote:
> > >
> > > > We are using Apache Parquet and Spark for a genome analysis platform,
> > > > called ADAM <http://bdgenomics.org>, that allows researchers to
> > quickly
> > > > analyze large datasets of DNA, RNA, etc. Parquet has been a core
> > > component
> > > > of the system and we see compression of ~20% compared to specialized
> > > genome
> > > > file formats e.g. compressed BAM. In short, we’re really happy with
> > > Parquet.
> > > >
> > > > We are using Avro Specific classes for almost all the entities in our
> > > > system, so Avro generates Java classes from our schema
> > > > <
> > >
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > > >.
> > > > Since the AvroIndexedRecordConverter has dictionary support, our
> > initial
> > > > load from disk to memory is compact.
> > > >
> > > > That’s the good news: compact on-disk and initial in-memory
> > > representation.
> > > >
> > > > Here’s the problem: the Spark shuffle.
> > > >
> > > > In order to integrate Parquet with Spark, we use a KryoRegistrator to
> > > > register Kryo serializers
> > > > <
> > >
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > > >
> > > > for each of our Avro objects (see Kryo Serializer interface
> > > > <
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > >).
> > > > We are serializing each object into record-oriented Avro, which makes
> > our
> > > > intermediate shuffle files much larger than the corresponding
> > > > column-oriented Parquet inputs. These large shuffle files are hurting
> > our
> > > > performance and limiting our scaling for some analysis.
> > > >
> > > > Since the shuffle data is short-lived, there’s no need store
> meta-data
> > > and
> > > > we have immediate access to schema through each Avro object. Each
> Avro
> > > > specific class has a SCHEMA$ field which contains the Avro Schema for
> > the
> > > > object. There are utility functions in parquet-avro which can convert
> > > > this Avro schema into Parquet schema. We also don’t need index pages,
> > > only
> > > > the dictionary and data pages. We don’t need predicates or projection
> > > > functionality. Does anyone on this list see a way to create a Parquet
> > > Kryo
> > > > serializer
> > > > <
> > >
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > > >
> > > > to read/write Avro Specific objects to/from a stream? Emitting
> > > > column-oriented data will understandably incur memory and CPU costs
> on
> > > the
> > > > map side but it will be worth it to improve our shuffle performance.
> > > >
> > > > This shuffle issue to slowing important research so any advice you
> have
> > > to
> > > > offer will be appreciated. Thank you.
> > > >
> > > > —
> > > > Matt Massie
> > > > UC Berkeley, AMPLab
> > > > 
> > > >
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Matt Massie <ma...@berkeley.edu>.

Thanks for the ideas, Julien. I appreciate it.

I suspected that I would need to direct my effort into a Parquet-based
Spark shuffle but wanted to reach out on the Parquet list first to make
sure that the simpler Parquet serialization approach wouldn't work.

As far as I can tell, there isn't an external API for extracting
dictionaries from Parquet files. Correct?

-Matt

On Tue, Apr 7, 2015 at 1:50 PM, Julien Le Dem <ju...@twitter.com.invalid>
wrote:

> You might want to look at Twitter Chill: https://github.com/twitter/chill
> This is where Twitter has the Kryo serializers for this kind of use cases.
> The one for Thrift:
>
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> Spark is using it already:
>
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
>
> But the main problem is that you want to encode multiple records at a time
> when the serializer gets a single key or value at a time.
> The output is not a single stream since values get routed to a different
> node based on the key.
>
> Unless spark first gets the list of values that goes to a specific node
> before encoding them (and provides an API to have custom encoding of all
> values together) I don't think you can implement a Parquet KryoSerializer.
> I would ask the Spark project whether it's possible to encode all the
> values that go to the same node together with a custom serializer.
>
> Another approach is to reuse dictionary encoding in your row oriented
> serialization.
> It would not compress as much as Parquet but may help. You'd need to pass
> the dictionary around but maybe you know all the possible values ahead of
> time. (extracting the dictionaries from the Parquet file?)
> After all your in-memory representation is compact because after dictionary
> decoding all avro struct that use the same string value point to the same
> java string instance, so you may be able to use this.
>
> I hope this helps
>
> Julien
>
>
>
> On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu> wrote:
>
> > I forgot to mention that I’m happy to do the work here, I’m mostly
> looking
> > for advice and pointers.
> >
> > Also, another interface that might be better for integration is the Spark
> > Serializer
> > <
> >
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> > >
> > since the SerializationStream and DeserializationStream have the flush()
> > and close() functions that are necessary (instead of just read/write).
> > 
> >
> > --
> > Matt <http://www.linkedin.com/in/mattmassie/> Massie
> > <http://www.twitter.com/matt_massie>
> > UC, Berkeley AMPLab <https://twitter.com/amplab>
> >
> > On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu>
> wrote:
> >
> > > We are using Apache Parquet and Spark for a genome analysis platform,
> > > called ADAM <http://bdgenomics.org>, that allows researchers to
> quickly
> > > analyze large datasets of DNA, RNA, etc. Parquet has been a core
> > component
> > > of the system and we see compression of ~20% compared to specialized
> > genome
> > > file formats e.g. compressed BAM. In short, we’re really happy with
> > Parquet.
> > >
> > > We are using Avro Specific classes for almost all the entities in our
> > > system, so Avro generates Java classes from our schema
> > > <
> >
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> > >.
> > > Since the AvroIndexedRecordConverter has dictionary support, our
> initial
> > > load from disk to memory is compact.
> > >
> > > That’s the good news: compact on-disk and initial in-memory
> > representation.
> > >
> > > Here’s the problem: the Spark shuffle.
> > >
> > > In order to integrate Parquet with Spark, we use a KryoRegistrator to
> > > register Kryo serializers
> > > <
> >
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> > >
> > > for each of our Avro objects (see Kryo Serializer interface
> > > <
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > >).
> > > We are serializing each object into record-oriented Avro, which makes
> our
> > > intermediate shuffle files much larger than the corresponding
> > > column-oriented Parquet inputs. These large shuffle files are hurting
> our
> > > performance and limiting our scaling for some analysis.
> > >
> > > Since the shuffle data is short-lived, there’s no need store meta-data
> > and
> > > we have immediate access to schema through each Avro object. Each Avro
> > > specific class has a SCHEMA$ field which contains the Avro Schema for
> the
> > > object. There are utility functions in parquet-avro which can convert
> > > this Avro schema into Parquet schema. We also don’t need index pages,
> > only
> > > the dictionary and data pages. We don’t need predicates or projection
> > > functionality. Does anyone on this list see a way to create a Parquet
> > Kryo
> > > serializer
> > > <
> >
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> > >
> > > to read/write Avro Specific objects to/from a stream? Emitting
> > > column-oriented data will understandably incur memory and CPU costs on
> > the
> > > map side but it will be worth it to improve our shuffle performance.
> > >
> > > This shuffle issue to slowing important research so any advice you have
> > to
> > > offer will be appreciated. Thank you.
> > >
> > > —
> > > Matt Massie
> > > UC Berkeley, AMPLab
> > > 
> > >
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Ryan Blue <bl...@cloudera.com>.

I did a hackathon project last year to get Spark working more smoothly 
with Avro, which you can take a look at here:

   https://github.com/rdblue/spark/commit/6e59f4d1

That fixes some cases with Avro records in Kryo where you have the 
schema on both ends and don't need to serialize the schema with every 
record, like specific and reflect. It also adds a typed stream, which 
works more like classic MR and allows you to set the schema for all of 
the records passing through a shuffle, without going through Kryo. This 
could save quite a bit of space, though not as much as Parquet would.

Julien is right that the main problem with using Parquet ends up being 
that you get benefits from encoding multiple records at once.

About the custom dictionaries, though, if you know which fields are 
getting good dictionary compression, you might be able to define an Avro 
schema that does the dictionary-encoding work via Avro enum. You could 
sample the data to find out the appropriate dictionary and generate the 
Avro schema that way.

rb

On 04/07/2015 01:50 PM, Julien Le Dem wrote:
> You might want to look at Twitter Chill: https://github.com/twitter/chill
> This is where Twitter has the Kryo serializers for this kind of use cases.
> The one for Thrift:
> https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
> Spark is using it already:
> https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320
>
> But the main problem is that you want to encode multiple records at a time
> when the serializer gets a single key or value at a time.
> The output is not a single stream since values get routed to a different
> node based on the key.
>
> Unless spark first gets the list of values that goes to a specific node
> before encoding them (and provides an API to have custom encoding of all
> values together) I don't think you can implement a Parquet KryoSerializer.
> I would ask the Spark project whether it's possible to encode all the
> values that go to the same node together with a custom serializer.
>
> Another approach is to reuse dictionary encoding in your row oriented
> serialization.
> It would not compress as much as Parquet but may help. You'd need to pass
> the dictionary around but maybe you know all the possible values ahead of
> time. (extracting the dictionaries from the Parquet file?)
> After all your in-memory representation is compact because after dictionary
> decoding all avro struct that use the same string value point to the same
> java string instance, so you may be able to use this.
>
> I hope this helps
>
> Julien
>
>
>
> On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu> wrote:
>
>> I forgot to mention that I’m happy to do the work here, I’m mostly looking
>> for advice and pointers.
>>
>> Also, another interface that might be better for integration is the Spark
>> Serializer
>> <
>> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
>>>
>> since the SerializationStream and DeserializationStream have the flush()
>> and close() functions that are necessary (instead of just read/write).
>> 
>>
>> --
>> Matt <http://www.linkedin.com/in/mattmassie/> Massie
>> <http://www.twitter.com/matt_massie>
>> UC, Berkeley AMPLab <https://twitter.com/amplab>
>>
>> On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu> wrote:
>>
>>> We are using Apache Parquet and Spark for a genome analysis platform,
>>> called ADAM <http://bdgenomics.org>, that allows researchers to quickly
>>> analyze large datasets of DNA, RNA, etc. Parquet has been a core
>> component
>>> of the system and we see compression of ~20% compared to specialized
>> genome
>>> file formats e.g. compressed BAM. In short, we’re really happy with
>> Parquet.
>>>
>>> We are using Avro Specific classes for almost all the entities in our
>>> system, so Avro generates Java classes from our schema
>>> <
>> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
>>> .
>>> Since the AvroIndexedRecordConverter has dictionary support, our initial
>>> load from disk to memory is compact.
>>>
>>> That’s the good news: compact on-disk and initial in-memory
>> representation.
>>>
>>> Here’s the problem: the Spark shuffle.
>>>
>>> In order to integrate Parquet with Spark, we use a KryoRegistrator to
>>> register Kryo serializers
>>> <
>> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
>>>
>>> for each of our Avro objects (see Kryo Serializer interface
>>> <
>> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
>>> ).
>>> We are serializing each object into record-oriented Avro, which makes our
>>> intermediate shuffle files much larger than the corresponding
>>> column-oriented Parquet inputs. These large shuffle files are hurting our
>>> performance and limiting our scaling for some analysis.
>>>
>>> Since the shuffle data is short-lived, there’s no need store meta-data
>> and
>>> we have immediate access to schema through each Avro object. Each Avro
>>> specific class has a SCHEMA$ field which contains the Avro Schema for the
>>> object. There are utility functions in parquet-avro which can convert
>>> this Avro schema into Parquet schema. We also don’t need index pages,
>> only
>>> the dictionary and data pages. We don’t need predicates or projection
>>> functionality. Does anyone on this list see a way to create a Parquet
>> Kryo
>>> serializer
>>> <
>> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
>>>
>>> to read/write Avro Specific objects to/from a stream? Emitting
>>> column-oriented data will understandably incur memory and CPU costs on
>> the
>>> map side but it will be worth it to improve our shuffle performance.
>>>
>>> This shuffle issue to slowing important research so any advice you have
>> to
>>> offer will be appreciated. Thank you.
>>>
>>> —
>>> Matt Massie
>>> UC Berkeley, AMPLab
>>> 
>>>
>>
>


-- 
Ryan Blue
Software Engineer
Cloudera, Inc.

Re: Doing the Spark shuffle on Parquet floors

Posted by Julien Le Dem <ju...@twitter.com.INVALID>.

You might want to look at Twitter Chill: https://github.com/twitter/chill
This is where Twitter has the Kryo serializers for this kind of use cases.
The one for Thrift:
https://github.com/twitter/chill/blob/develop/chill-thrift/src/main/java/com/twitter/chill/thrift/TBaseSerializer.java
Spark is using it already:
https://github.com/apache/spark/blob/ee11be258251adf900680927ba200bf46512cc04/pom.xml#L320

But the main problem is that you want to encode multiple records at a time
when the serializer gets a single key or value at a time.
The output is not a single stream since values get routed to a different
node based on the key.

Unless spark first gets the list of values that goes to a specific node
before encoding them (and provides an API to have custom encoding of all
values together) I don't think you can implement a Parquet KryoSerializer.
I would ask the Spark project whether it's possible to encode all the
values that go to the same node together with a custom serializer.

Another approach is to reuse dictionary encoding in your row oriented
serialization.
It would not compress as much as Parquet but may help. You'd need to pass
the dictionary around but maybe you know all the possible values ahead of
time. (extracting the dictionaries from the Parquet file?)
After all your in-memory representation is compact because after dictionary
decoding all avro struct that use the same string value point to the same
java string instance, so you may be able to use this.

I hope this helps

Julien

On Tue, Apr 7, 2015 at 12:07 PM, Matt Massie <ma...@berkeley.edu> wrote:

> I forgot to mention that I’m happy to do the work here, I’m mostly looking
> for advice and pointers.
>
> Also, another interface that might be better for integration is the Spark
> Serializer
> <
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala
> >
> since the SerializationStream and DeserializationStream have the flush()
> and close() functions that are necessary (instead of just read/write).
> 
>
> --
> Matt <http://www.linkedin.com/in/mattmassie/> Massie
> <http://www.twitter.com/matt_massie>
> UC, Berkeley AMPLab <https://twitter.com/amplab>
>
> On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu> wrote:
>
> > We are using Apache Parquet and Spark for a genome analysis platform,
> > called ADAM <http://bdgenomics.org>, that allows researchers to quickly
> > analyze large datasets of DNA, RNA, etc. Parquet has been a core
> component
> > of the system and we see compression of ~20% compared to specialized
> genome
> > file formats e.g. compressed BAM. In short, we’re really happy with
> Parquet.
> >
> > We are using Avro Specific classes for almost all the entities in our
> > system, so Avro generates Java classes from our schema
> > <
> https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl
> >.
> > Since the AvroIndexedRecordConverter has dictionary support, our initial
> > load from disk to memory is compact.
> >
> > That’s the good news: compact on-disk and initial in-memory
> representation.
> >
> > Here’s the problem: the Spark shuffle.
> >
> > In order to integrate Parquet with Spark, we use a KryoRegistrator to
> > register Kryo serializers
> > <
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala
> >
> > for each of our Avro objects (see Kryo Serializer interface
> > <
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> >).
> > We are serializing each object into record-oriented Avro, which makes our
> > intermediate shuffle files much larger than the corresponding
> > column-oriented Parquet inputs. These large shuffle files are hurting our
> > performance and limiting our scaling for some analysis.
> >
> > Since the shuffle data is short-lived, there’s no need store meta-data
> and
> > we have immediate access to schema through each Avro object. Each Avro
> > specific class has a SCHEMA$ field which contains the Avro Schema for the
> > object. There are utility functions in parquet-avro which can convert
> > this Avro schema into Parquet schema. We also don’t need index pages,
> only
> > the dictionary and data pages. We don’t need predicates or projection
> > functionality. Does anyone on this list see a way to create a Parquet
> Kryo
> > serializer
> > <
> https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java
> >
> > to read/write Avro Specific objects to/from a stream? Emitting
> > column-oriented data will understandably incur memory and CPU costs on
> the
> > map side but it will be worth it to improve our shuffle performance.
> >
> > This shuffle issue to slowing important research so any advice you have
> to
> > offer will be appreciated. Thank you.
> >
> > —
> > Matt Massie
> > UC Berkeley, AMPLab
> > 
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Matt Massie <ma...@berkeley.edu>.

I forgot to mention that I’m happy to do the work here, I’m mostly looking
for advice and pointers.

Also, another interface that might be better for integration is the Spark
Serializer
<https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala>
since the SerializationStream and DeserializationStream have the flush()
and close() functions that are necessary (instead of just read/write).


--
Matt <http://www.linkedin.com/in/mattmassie/> Massie
<http://www.twitter.com/matt_massie>
UC, Berkeley AMPLab <https://twitter.com/amplab>

On Tue, Apr 7, 2015 at 11:49 AM, Matt Massie <ma...@berkeley.edu> wrote:

> We are using Apache Parquet and Spark for a genome analysis platform,
> called ADAM <http://bdgenomics.org>, that allows researchers to quickly
> analyze large datasets of DNA, RNA, etc. Parquet has been a core component
> of the system and we see compression of ~20% compared to specialized genome
> file formats e.g. compressed BAM. In short, we’re really happy with Parquet.
>
> We are using Avro Specific classes for almost all the entities in our
> system, so Avro generates Java classes from our schema
> <https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl>.
> Since the AvroIndexedRecordConverter has dictionary support, our initial
> load from disk to memory is compact.
>
> That’s the good news: compact on-disk and initial in-memory representation.
>
> Here’s the problem: the Spark shuffle.
>
> In order to integrate Parquet with Spark, we use a KryoRegistrator to
> register Kryo serializers
> <https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala>
> for each of our Avro objects (see Kryo Serializer interface
> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>).
> We are serializing each object into record-oriented Avro, which makes our
> intermediate shuffle files much larger than the corresponding
> column-oriented Parquet inputs. These large shuffle files are hurting our
> performance and limiting our scaling for some analysis.
>
> Since the shuffle data is short-lived, there’s no need store meta-data and
> we have immediate access to schema through each Avro object. Each Avro
> specific class has a SCHEMA$ field which contains the Avro Schema for the
> object. There are utility functions in parquet-avro which can convert
> this Avro schema into Parquet schema. We also don’t need index pages, only
> the dictionary and data pages. We don’t need predicates or projection
> functionality. Does anyone on this list see a way to create a Parquet Kryo
> serializer
> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>
> to read/write Avro Specific objects to/from a stream? Emitting
> column-oriented data will understandably incur memory and CPU costs on the
> map side but it will be worth it to improve our shuffle performance.
>
> This shuffle issue to slowing important research so any advice you have to
> offer will be appreciated. Thank you.
>
> —
> Matt Massie
> UC Berkeley, AMPLab
> 
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Julien Le Dem <ju...@twitter.com.INVALID>.

or even better, we retweet you :)

On Wed, Apr 8, 2015 at 2:00 PM, Matt Massie <ma...@berkeley.edu> wrote:

> Sure, you can tweet that.
>
> --
> Matt <http://www.linkedin.com/in/mattmassie/> Massie
> <http://www.twitter.com/matt_massie>
> UC, Berkeley AMPLab <https://twitter.com/amplab>
>
> On Wed, Apr 8, 2015 at 1:54 PM, Ryan Blue <bl...@cloudera.com> wrote:
>
> > "Parquet has been a core component of the system and we see compression
> of
> > ~20% compared to specialized genome file formats e.g. compressed BAM. In
> > short, we’re really happy with Parquet."
> >
> > Matt, can we tweet this? That's great!
> >
> > rb
> >
> > On 04/07/2015 11:49 AM, Matt Massie wrote:
> >
> >> We are using Apache Parquet and Spark for a genome analysis platform,
> >> called ADAM <http://bdgenomics.org>, that allows researchers to quickly
> >> analyze large datasets of DNA, RNA, etc. Parquet has been a core
> component
> >> of the system and we see compression of ~20% compared to specialized
> >> genome
> >> file formats e.g. compressed BAM. In short, we’re really happy with
> >> Parquet.
> >>
> >> We are using Avro Specific classes for almost all the entities in our
> >> system, so Avro generates Java classes from our schema
> >> <https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/
> >> resources/avro/bdg.avdl>.
> >> Since the AvroIndexedRecordConverter has dictionary support, our initial
> >> load from disk to memory is compact.
> >>
> >> That’s the good news: compact on-disk and initial in-memory
> >> representation.
> >>
> >> Here’s the problem: the Spark shuffle.
> >>
> >> In order to integrate Parquet with Spark, we use a KryoRegistrator to
> >> register Kryo serializers
> >> <
> https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/
> >> scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala>
> >> for each of our Avro objects (see Kryo Serializer interface
> >> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/
> >> esotericsoftware/kryo/Serializer.java>).
> >> We are serializing each object into record-oriented Avro, which makes
> our
> >> intermediate shuffle files much larger than the corresponding
> >> column-oriented Parquet inputs. These large shuffle files are hurting
> our
> >> performance and limiting our scaling for some analysis.
> >>
> >> Since the shuffle data is short-lived, there’s no need store meta-data
> and
> >> we have immediate access to schema through each Avro object. Each Avro
> >> specific class has a SCHEMA$ field which contains the Avro Schema for
> the
> >> object. There are utility functions in parquet-avro which can convert
> this
> >> Avro schema into Parquet schema. We also don’t need index pages, only
> the
> >> dictionary and data pages. We don’t need predicates or projection
> >> functionality. Does anyone on this list see a way to create a Parquet
> Kryo
> >> serializer
> >> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/
> >> esotericsoftware/kryo/Serializer.java>
> >> to read/write Avro Specific objects to/from a stream? Emitting
> >> column-oriented data will understandably incur memory and CPU costs on
> the
> >> map side but it will be worth it to improve our shuffle performance.
> >>
> >> This shuffle issue to slowing important research so any advice you have
> to
> >> offer will be appreciated. Thank you.
> >>
> >> —
> >> Matt Massie
> >> UC Berkeley, AMPLab
> >> 
> >>
> >>
> >
> > --
> > Ryan Blue
> > Software Engineer
> > Cloudera, Inc.
> >
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Matt Massie <ma...@berkeley.edu>.

Sure, you can tweet that.

--
Matt <http://www.linkedin.com/in/mattmassie/> Massie
<http://www.twitter.com/matt_massie>
UC, Berkeley AMPLab <https://twitter.com/amplab>

On Wed, Apr 8, 2015 at 1:54 PM, Ryan Blue <bl...@cloudera.com> wrote:

> "Parquet has been a core component of the system and we see compression of
> ~20% compared to specialized genome file formats e.g. compressed BAM. In
> short, we’re really happy with Parquet."
>
> Matt, can we tweet this? That's great!
>
> rb
>
> On 04/07/2015 11:49 AM, Matt Massie wrote:
>
>> We are using Apache Parquet and Spark for a genome analysis platform,
>> called ADAM <http://bdgenomics.org>, that allows researchers to quickly
>> analyze large datasets of DNA, RNA, etc. Parquet has been a core component
>> of the system and we see compression of ~20% compared to specialized
>> genome
>> file formats e.g. compressed BAM. In short, we’re really happy with
>> Parquet.
>>
>> We are using Avro Specific classes for almost all the entities in our
>> system, so Avro generates Java classes from our schema
>> <https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/
>> resources/avro/bdg.avdl>.
>> Since the AvroIndexedRecordConverter has dictionary support, our initial
>> load from disk to memory is compact.
>>
>> That’s the good news: compact on-disk and initial in-memory
>> representation.
>>
>> Here’s the problem: the Spark shuffle.
>>
>> In order to integrate Parquet with Spark, we use a KryoRegistrator to
>> register Kryo serializers
>> <https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/
>> scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala>
>> for each of our Avro objects (see Kryo Serializer interface
>> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/
>> esotericsoftware/kryo/Serializer.java>).
>> We are serializing each object into record-oriented Avro, which makes our
>> intermediate shuffle files much larger than the corresponding
>> column-oriented Parquet inputs. These large shuffle files are hurting our
>> performance and limiting our scaling for some analysis.
>>
>> Since the shuffle data is short-lived, there’s no need store meta-data and
>> we have immediate access to schema through each Avro object. Each Avro
>> specific class has a SCHEMA$ field which contains the Avro Schema for the
>> object. There are utility functions in parquet-avro which can convert this
>> Avro schema into Parquet schema. We also don’t need index pages, only the
>> dictionary and data pages. We don’t need predicates or projection
>> functionality. Does anyone on this list see a way to create a Parquet Kryo
>> serializer
>> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/
>> esotericsoftware/kryo/Serializer.java>
>> to read/write Avro Specific objects to/from a stream? Emitting
>> column-oriented data will understandably incur memory and CPU costs on the
>> map side but it will be worth it to improve our shuffle performance.
>>
>> This shuffle issue to slowing important research so any advice you have to
>> offer will be appreciated. Thank you.
>>
>> —
>> Matt Massie
>> UC Berkeley, AMPLab
>> 
>>
>>
>
> --
> Ryan Blue
> Software Engineer
> Cloudera, Inc.
>

Re: Doing the Spark shuffle on Parquet floors

Posted by Ryan Blue <bl...@cloudera.com>.

"Parquet has been a core component of the system and we see compression 
of ~20% compared to specialized genome file formats e.g. compressed BAM. 
In short, we’re really happy with Parquet."

Matt, can we tweet this? That's great!

rb

On 04/07/2015 11:49 AM, Matt Massie wrote:
> We are using Apache Parquet and Spark for a genome analysis platform,
> called ADAM <http://bdgenomics.org>, that allows researchers to quickly
> analyze large datasets of DNA, RNA, etc. Parquet has been a core component
> of the system and we see compression of ~20% compared to specialized genome
> file formats e.g. compressed BAM. In short, we’re really happy with Parquet.
>
> We are using Avro Specific classes for almost all the entities in our
> system, so Avro generates Java classes from our schema
> <https://github.com/bigdatagenomics/bdg-formats/blob/master/src/main/resources/avro/bdg.avdl>.
> Since the AvroIndexedRecordConverter has dictionary support, our initial
> load from disk to memory is compact.
>
> That’s the good news: compact on-disk and initial in-memory representation.
>
> Here’s the problem: the Spark shuffle.
>
> In order to integrate Parquet with Spark, we use a KryoRegistrator to
> register Kryo serializers
> <https://github.com/bigdatagenomics/adam/blob/master/adam-core/src/main/scala/org/bdgenomics/adam/serialization/ADAMKryoRegistrator.scala>
> for each of our Avro objects (see Kryo Serializer interface
> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>).
> We are serializing each object into record-oriented Avro, which makes our
> intermediate shuffle files much larger than the corresponding
> column-oriented Parquet inputs. These large shuffle files are hurting our
> performance and limiting our scaling for some analysis.
>
> Since the shuffle data is short-lived, there’s no need store meta-data and
> we have immediate access to schema through each Avro object. Each Avro
> specific class has a SCHEMA$ field which contains the Avro Schema for the
> object. There are utility functions in parquet-avro which can convert this
> Avro schema into Parquet schema. We also don’t need index pages, only the
> dictionary and data pages. We don’t need predicates or projection
> functionality. Does anyone on this list see a way to create a Parquet Kryo
> serializer
> <https://github.com/EsotericSoftware/kryo/blob/master/src/com/esotericsoftware/kryo/Serializer.java>
> to read/write Avro Specific objects to/from a stream? Emitting
> column-oriented data will understandably incur memory and CPU costs on the
> map side but it will be worth it to improve our shuffle performance.
>
> This shuffle issue to slowing important research so any advice you have to
> offer will be appreciated. Thank you.
>
> —
> Matt Massie
> UC Berkeley, AMPLab
> 
>


-- 
Ryan Blue
Software Engineer
Cloudera, Inc.