How do you deserialize Kryo in classes using Scalding?

Question

How do you deserialize Kryo in classes using Scalding?

I know that Scalding serialization uses Kryo by default. So, for this example, let's say I have a pipe of student objects.

case class Student(name:String, id:String)

val pipe: Pipe[Student] = //....

Then I write this pipe to a TextDelimited file using Kryo.

pipe.write(args("output"))

Now that I have these “output” files, how can I easily rewrite them into Student objects? My general idea is something like the following, but that doesn't work.

Tsv("Kryo output files").read.map {bytes => 
  val instantiator = (new ScalaKryoInstantiator).setRegistrationRequired(true)
  val kryo = instantiator.newKryo

  kryo.register(classOf[Person])

  val deserialized = kryo.readObject(new Input(bytes), classOf[Person])

  deserialized
}