Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Optimizing KyroCoder in beam backend #1955

Open
nownikhil opened this issue Oct 13, 2021 · 7 comments
Open

Optimizing KyroCoder in beam backend #1955

nownikhil opened this issue Oct 13, 2021 · 7 comments

Comments

@nownikhil
Copy link
Contributor

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

)
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 (

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
Copy link
Collaborator

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
Copy link
Collaborator

ianoc commented Oct 13, 2021

#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
Copy link
Collaborator

yeah, maybe here:

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
Copy link
Collaborator

ianoc commented Oct 14, 2021

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

final def run(conf: Config, mode: Mode)(implicit cec: ConcurrentExecutionContext): Future[T] = {
for registering should be good i think?

@johnynek
Copy link
Collaborator

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
Copy link
Collaborator

ianoc commented Oct 14, 2021

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
Copy link
Collaborator

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

3 participants