Add support for DataSets of Avro records

[Zariel]

It would be very useful to be able to load an Avro DataSet and have the specific records, eg

val ds = sqlContext.read.load("path").as[SomeAvroType]

[bdrillard]

I've prepared a solution for this request. I've added an object, AvroEncoder, which produces an SQL Encoder for Avro objects of a given Schema or class, and supports creation of DataSets for both SpecificRecord and GenericData.Record. We can create DataSets of objects that implement IndexedRecord by calling one of the following methods:

val encoder = AvroEncoder.of[SomeAvroClass](classOf[SomeAvroClass]) // for SpecificRecord
val encoder = AvroEncoder.of[SomeAvroClass](someAvroSchema) // an alternative using the class schema
val encoder = AvroEncoder.of[GenericData.Record](someAvroSchema) // for GenericData.Record

Then we can provide the encoder as the argument of as:

val ds = sqlContext.read.load("path").as(encoder)

There's a caveat though: my implementation depends on the ObjectType class in Catalyst, which is currently private. In the mailing list the Spark folks said they'd entertain a pull-request to make the class public. In the mean time, I add a packaged jar of Catalyst from my own fork of the Spark branch-2.0 repo to the classpath of my spark-avro build, and the AvroEncoder works as expected.

Also there's no current implementation for the Avro Map type yet. Catalyst has a function ExternalMapToCatalyst that could probably be used to provide that functionality, but it's not available in Spark 2.0.x yet, so I wasn't able to build with it.

All the other Avro types serialize/deserialize though! Hope you find this useful in the interim.

[JoshRosen]

@bdrillard, given that ObjectType now appears to be public in Spark, would you be interested in submitting a pull request to contribute your custom encoder to this project?

[JoshRosen]

Per my comment at #89 (comment), this is a feature that we absolutely should add, so I'll gladly review any patches for this and will help to carry them across the finish line.

[bdrillard]

@JoshRosen, I'd be glad to create a pull-request for this issue. However, SPARK-17770 to make the ObjectType public was merged into the as-of-yet unreleased Spark-2.1.0 branch and not into the current 2.0.2 release. In a pull request, I'm not sure which Spark-SQL version I'd be able to reference for SBT that would have the ObjectType properly accessible.

[JoshRosen]

@bdrillard, although it would be ugly, there's a possibility that we use an ugly hack to either construct ObjectType reflectively or to place a helper "shim" class into the org.apache.spark.sql namespace in order to make parts of ObjectType visible to this library from behind a shim. This is going to be extremely ugly but I think it would work with Spark 2.0.x and we could test it against all of the patch releases to make sure that it's compatible.

Here's my short-term suggestion: submit what you have now as a PR and update the .travis.yml file and build.sbt files to build against 2.1.0-SNAPSHOT; once we validate that things work with ObjectType exposed then I can help to add the reflection hacks to make this backwards-compatible.

[bdrillard]

Working on uplifting the Spark dependency to 2.1.0-SNAPSHOT, however, it appears a recent pull-request for SPARK-18024 has added a pair of new abstract methods to the OutputWriter class and made some changes apparently in how folders are read.

I've tried to update the implementation of AvroOutputWriterFactory to reflect the change, but now any test that reads from a temp directory where some avro data had been written fails with

java.io.FileNotFoundException: No Avro files found. Hadoop option "avro.mapred.ignore.inputs.without.extension" is set to true. Do all input files have ".avro" extension?

@rxin, since you made the pull request for SPARK-18024, could you tell us, is there an additional step that needs to be taken here to uplift the AvroOutputWriterFactory to the new API?

The remaining unit-tests I had added for the encoder pass as normal still.

[Himanshu-parmar13]

I want to use this project to encode my avro data in Parquet file using Java. Can you please provide example for the same.

[bdrillard]

I've prepared a branch to create a pull-request from that would make use of the uplift patch done in 206. Once that PR is merged, I can create a separate PR for the AvroEncoder. Thanks @koertkuipers for figuring out the upgrade!