Avro revisited

[Christopher Severs]

I love that Scoobi has Avro support but it doesn't allow me to do what I want to do so I added some new methods.

Namely, I want to read up records into either a GenericData.Record or some class implementing SpecificRecord and then do stuff with them. This way I'm not limited to having only 8 fields. I would also like to be able to write records with whatever names and field names I want. So far I have the read part done and working. I don't think the write part will be much more than a bit of copy/paste.

The changes are here:
https://github.com/ccsevers/scoobi/blob/master/src/main/scala/com/nicta/scoobi/io/avro/AvroInput.scala

The new methods are fromAvroGeneric and fromAvroSpecific. They look almost identical to the fromAvroFile but return a DList[T] where T <: GenericRecord or T<:SpecificRecord. They use the standard avro serialization for the wireformat. This is actually where I think things might be improved. The current toWire is a bit ugly:

	    	def toWire(x: T, out: DataOutput) { 
	      val schema : Schema = x.getSchema
	      val byteStream : ByteArrayOutputStream  = new ByteArrayOutputStream
	      val encoder : Encoder = EncoderFactory.get.binaryEncoder(byteStream, null)
	      val datumWriter : GenericDatumWriter[T] = new GenericDatumWriter[T](schema)
	      val dataFileWriter : DataFileWriter[T] = new DataFileWriter[T]( datumWriter)
	      dataFileWriter.create(schema, byteStream);
	      byteStream.flush
	      val byteArray = byteStream.toByteArray
	      out.writeInt(byteArray.size)
	      out.write(byteArray) 
	    }

It seems like there should be a better way to do this, but the Avro writers take things which inherit from OutputStream and the toWire uses DataOutput. They seem to intersect in something like DataOutputStream but I'm not sure that will work.

I tried to do this in the framework of AvroSchema but I didn't see a clean way to get at val schema without having an object in hand. I'll keep playing with this but if anyone sees a better way to do things please let me know. The end goal is to also be able to take in any class and use the reflect API in avro to read in or out avro files.

Regards,
Chris

[Ben Lever]

Hi Chris,

Apologies for a slow response.

When you say "not limited to having only 8 fields", where does the 8 come from?

As for supporting GenericData.Record and SpecificRecord, it would be good if you could leverage the existing toAvroFile and fromAvroFile by simply creating AvroSchemas. I think you were eluding to that as a solution but were unsure how to get hold of a GenericData.Record so you could get its Schema? I think you should be able to create an AvroSchema type class instance for a specific GenericData.Record object. Let me know if you need a hand with that.

I don't think we need new methods like you suggest, but maybe what would be useful are methods that take an explicit AvroSchema (rather than implicitly) - e.g. fromAvroFileWithSchema and toAvroFileWithSchema (or maybe we could simply overload fromAvroFile and toAvroFile).

Cheers,

Ben.

[Christopher Severs]

Hi Ben,

No worries, I was out on vacation all last week anyway.

For the 8 fields, as far as I can tell, in the AvroSchema.scala code the largest number of fields that can be handled is in Tuple8Schema which reads in a GenericData.Record, gets the first 8 fields and then outputs a Tuple of those fields. My use case is generally reading in records with more than 8 fields and often using the schema to do some projection and only get the fields I want. Both of these capabilities would be nice to have and it would be simple to extend the number of fields handled but at some point we also bump into the max size for a tuple too.

Similarly for the output I would like to have some finer control (be able to name fields and add other info) over what the output record looks like. Having the schema in hand helps here as well.

I agree the best solution would be use the existing AvroSchema code but as you noticed I wasn't sure how to set the schema without taking it in explicitly. Something like the below would be nice but I don't see a good way to populate the schema:

implicit def GenericDataRecordSchema = new AvroSchema[GenericData.Record] {

    type AvroType = GenericData.Record

    val schema: Schema = ???

    def fromAvro(x: GenericData.Record): GenericData.Record = x

    def toAvro(x: GenericData.Record): GenericData.Record = x

  }

If you (or anyone else) sees a nice way to do this I think most things would be solved immediately.

If that's not possible then overloading the current fromAvroFile/toAvroFile would be fine with me. I mainly broke them into different methods for ease of reading.

I think also with the newest version of Avro I can use SpecificData instead and it will fall back on GenericData when it needs to. I'll look into that since it would save having multiple methods.

Thanks for the help,
Chris

[Ben Lever]

Hi Chris,

Thanks for clarifying the 8 fields - can't believe I didn't make the connection myself :)

In terms of how you would implement an AvroSchema type class instance I was imagining something like the following:

def fromSchema(sch: Schema): AvroSchema[GenericData.Record] = new AvroSchema[GenericData.Record] {

    type AvroType = GenericData.Record

    val schema: Schema = sch

    def fromAvro(x: GenericData.Record): GenericData.Record = x

    def toAvro(x: GenericData.Record): GenericData.Record = x

}

val mySpecificSchema: Schema = ... // has all the correct fields you want, etc

implicit val myAvroSchema = fromSchema(mySpecificSchema)

Note that I haven't checked if this compiles let alone works. Would be interested to see if something like this works for you though. If it does, I think it's a good starting point for generalising the Avro I/O APIs to better support existing/specific schemas.

Cheers,

Ben.

[Christopher Severs]

Hi Ben,

I'll give this a shot and let you know. I'll also be at Hadoop Summit and the meetup on the 11th so maybe we can find a time to chat then as well. I can relay any of Alex's comments too.

Regards,
Chris

[Christopher Severs]

I gave this some more time and have a nice solution. I fit Avro Generic/SpecificRecord support into the regular AvroSchema framework so there is no need for different methods. I opened a pull request on the cdh3 branch. There is an example of usage included too.

----
Chris

[Ben Wing]

Hey, I just noticed this; I forgot about scoobi-dev.

I wrote some code to auto-generate the necessary WireFormat implicits for case classes up to 22 args, since I kept running up against the previous 8-argument limit. (Scala only supports tuples and case classes with up to 22 members.  Functions can have more than this but this isn't too helpful for Scoobi, I don't think.) This has been merged into the latest Scoobi.

The same code could be modified easily to handle up to 22 fields in AvroSchemas.  In the long run this might not be enough for you, but in the short run it might work fine, and should be easy to get incorporated into Scoobi.

ben

[Christopher Severs]

Hi Ben,

That would definitely be helpful but ideally I would like more standard Avro support. The pull request I made recently adds the ability for scoobi to handle Avro Specific and Generic Records directly. This solves all my use cases and puts it on par with regular mapreduce avro support.

The next step would be to get it to work with the avro compiler plugin here:
https://github.com/radlab/SCADS/wiki/Avro-Plugin

The plugin allows you to define a case class and have it extend their AvroRecord and it does all the rest (making sure it implements Avro SpecificRecord and doing type conversions). The issue I'm having is that a couple Avro pieces call a public field, SCHEMA$, which isn't present in the interface or abstract class that specific records implement. There doesn't seem to be a way to have a public field in Scala so this creates a problem. I've tried changing Avro to use the getSchema method instead (via class.newInstance().getSchema() or via getMethod...) but that breaks some things elsewhere in Avro.

Thanks,
Chris

[Ben Wing]

Hey Chris ...

I'm not a Scoobi developer so I can't speak to getting your pull request in.  Generally I've found them quite receptive and helpful, although it seems to depend on how big the change is ...  I think there's more resistance especially to changes that expand the public API, for understandable reasons.

On Monday, October 1, 2012 3:46:54 PM UTC-4, Christopher Severs wrote:

Hi Ben,

That would definitely be helpful but ideally I would like more standard Avro support. The pull request I made recently adds the ability for scoobi to handle Avro Specific and Generic Records directly. This solves all my use cases and puts it on par with regular mapreduce avro support.

The next step would be to get it to work with the avro compiler plugin here:
https://github.com/radlab/SCADS/wiki/Avro-Plugin

The plugin allows you to define a case class and have it extend their AvroRecord and it does all the rest (making sure it implements Avro SpecificRecord and doing type conversions). The issue I'm having is that a couple Avro pieces call a public field, SCHEMA$, which isn't present in the interface or abstract class that specific records implement. There doesn't seem to be a way to have a public field in Scala so this creates a problem. I've tried changing Avro to use the getSchema method instead (via class.newInstance().getSchema() or via getMethod...) but that breaks some things elsewhere in Avro.

I imagine that the inability to define a public field in Scala is forced by the lack of multiple inheritance in Java.  When you have a child class that inherits from a Scala trait and the trait defines a method, Scala can implement this by adding a method definition in the child class that calls the trait's method; but there's no way to do something similar for non-private fields.

Is it possible for you to simply add a bit of Java adapter code to handle the public field?

ben

[Christopher Severs]

Hi Ben,

I assume they're a bit busy and will need some time to vet and/or rewrite the changes. I thought it would be better to contribute a working version of what I'm after rather than asking someone else to do it. I can run my own branch for the moment and it works fine.

On the public field issue I (well, a colleague and I) tried making a java abstract class with that field defined and having it call the getSchema function to set it and then had the scala class implement that class in addition to the AvroRecord. Oddly, it compiled fine but when you query it for the SCHEMA$ field it throws a runtime exception. I think it's really a fundamental issue/flaw in how Avro specific is implemented. They have an interface and abstract class that you are supposed to use but neither of those mention this magic field (but the interface does have a getSchema method). I had hoped it would be a simple change to using the getSchema function but there must be some other things going on under the hood that I haven't tracked down yet and which make it not so trivial to change.

Appreciate the help,
Chris

[Ben Lever]

Hi Chris,

Thanks for you pull request. I've only had a quick skim at this stage but will hopefully have a proper look within the week - sorry for the delay. It sounds like it's solving the particular problem you were telling me about ~6 months ago ... do you think there's anything missing that you might eventually need?

Cheers,

Ben.

[Christopher Severs]

Hi Ben,

No worries. Yes, it solves the issue we talked about way back when. Rather than redo everything from scratch I found a way to use the existing AvroScheme approach and just added a new implicit there for things which implement Avro GenericContainer. I also added an implicit wireformat using the Avro IO pieces but my guess is that it could be much better. 

We also talked a bit about code gen for Avro classes and there is a very nice project here:

https://github.com/radlab/SCADS/wiki/Avro-Plugin

As I noted above there is some oddness with Avro itself which is causing me grief in making this work with Scoobi. I think I'll go straight to the Avro folks and see if they can help out with that part. If that works then we should be all set. 

Regards,

Chris