Optimizing KyroCoder in beam backend

[nownikhil]

Current implementation of KryoCoder writes class for every object on the output stream. (

scalding/scalding-beam/src/main/scala/com/twitter/scalding/beam_backend/KryoCoder.scala

Line 16 in b0ba993

val bytes = kryoPool.toBytesWithClass(value)

)
This was done because beam can split the stream in between and if registration is only in the beginning of the stream, the latter part of the stream will fail. However we don't want to write className for classes which are already registered.

We can set setRegistrationRequired(true) when creating the Instantiator (

scalding/scalding-beam/src/main/scala/com/twitter/scalding/beam_backend/BeamBackend.scala

Line 22 in b0ba993

implicit val kryoCoder: KryoCoder = new KryoCoder(defaultKryoCoderConfiguration(config))

).

Then in KryoCoder we can keep a mapping of classes which have registration available (We can do a Try {pool.hasRegistration} and save the output in a map for future) and for those we use kryoPool.toBytesWithoutClass and for others we do kryoPool.toBytesWithClass

Is there a better way to achieve this?

[johnynek]

we can't really use toBytesWithoutClass unless we know for sure the class. Note, that is different from knowing the type.

In Hadoop, it can give you the class in some cases (and cascading also has a mechanism for this, which is why it was there).

Note: kryo will write a registration number instead of a classname when you do toBytesWithClass if the class type is registered, which is why we allow setting up the kyroinstantitator in the config. We try to register most common scala standard library classes, but also you can (should) set up a twitter kryoinstantiator that knows about thrift types in play.

We had some idea in the past of using scala reflection to try to enumerate the possible classes in play in a job and register those, but I don't know what the state of that was. I don't recall where that code lived.

cc @ianoc who may have some recollections.

[ianoc]

#1654

Thats the PR that added it , I'm not sure how well it would work with the execution API though. the test is only aimed at Job rather than execution.

The optimizer/planner could probably add a phase to try use this reflection and traverse an execution to build it up. (There would be gremlins to be careful around only doing it in a means that can't result in cached data between jobs/flatmaped executions getting invalid).

[johnynek]

yeah, maybe here:

scalding/scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

Line 766 in b0ba993

def execute(conf: Config, writes: List[ToWrite[_]])(implicit

(and in the other two: toIterable and force) we could traverse the typedpipes you have that you are about to run, update the config with new classnames, and it might "just work"?

[ianoc]

Yep - ensuring zipped executions share a synchronized registry can probably be setup on a per-execution run basis here: ?

scalding/scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

Line 151 in b0ba993

final def run(conf: Config, mode: Mode)(implicit cec: ConcurrentExecutionContext): Future[T] = {

for registering should be good i think?

[johnynek]

Okay, you make a good point: you can't change the registration between stages because you need to be able to deserialize data from earlier stages in later stages. This basically means that you can't see inside flatmaps, which I think is what you are implying.

I think the only safe place to put the token updating is where you linked: just before you run. You should walk the static value you have inside def run and that's that. That may mean in practice a lot gets hidden since flatMap is used internally.

That said... I have always been a bit skeptical this will be a giant win since I think we almost always compress output and the classnames will be basically the first things to be compressed, but I could be wrong (it may be that allocating all those strings is a waste of memory if Class does not cache the name allocation.

[ianoc]

I think the requirement I was thinking of is a bit looser, with the logic:

1. If a type is registered now but wasn't earlier, then the classname is inline, so it should work
2. Thus we can add types in later
3. However, we must always have a agreed upon adding order

Which i think a global lock for an execution would work? or mutex guarded state around the class registration?

(1) should hold right?

Its been many many years since i've ever benchmarked/profiled this, but i could believe at least that the internode speeds aren't often the determining factor now. The serializing of the string many many times could be a big factor, along with flushing buffers to disk super often. If you take skewed data or reduction to a single reducer/hot join key -- in these cases the serialization/deserialization of the classname, compression of data cpu time and disk I/O i believe would be pretty meaningful?

[johnynek]

yeah, I think you are right Ian: you could before you run any stage potentially update a dynamic map as long as it has a mutex. So you only add to the map of class to numbers. That should work.