MOUT not properly handled

[ivanprado]

Current MOUT is not acting as a total OutputFormat. MOUT life cycle is managed by calls in the methods setup(), write() and cleanup() from mappers and reducers. Ant the commit is handled at the cleanup() method. That is not right. It should be handled by an OutputCommiter.

By the other hand, the responsible for commiting the results, including the MOUT results, is the principal OutputFormat. That means that MOUT would only work if the OutputFormat extends FileOutputFormat, which is not right. MOUT should work independently of the principal OutputFormat.

The solution should be capturing the principal OutputFormat signals by using some kind of ProxyOutputFormta and dealing with these signals to the different OutputFormat, the principal and the MOUT. Also we should have some kind of ProxyOutputCommiter, that should send signals properly to both output systems.

[pereferrera]

Just as a remainder, Pangool already has a ProxyOutputFormat which is the principal OutputFormat of any Pangool Job. So I'm not sure about your second and third paragraph being right. If you take a look to PangoolMultipleOutputs.getRecordWriter() you'll see that the OutputCommitter is a ProxyOutputCommitter.

And just to remember something; Pangool MultipleOutputs behave exactly like Hadoop MultipleOutputs, only that they are more flexible. Hadoop MultipleOutputs also handles the lifecycle like that (setup(), write() and cleanup()).

This doesn't mean we shouldn't make this better. In the end Pangool is about making things better than the raw Hadoop Java API :-)

But I'm now just wondering what is the issue exactly? What is or could be the problem with how this is implemented now? Can you take a deeper look and ellaborate more on the problems of the current implementation?

[ivanprado]

This topic is complex, so I'm try to do my bests explaining what I understand.

Problem

1. The commit process is something that is initiated by Hadoop. The OutputFormat must provide an OutputCommiter through the getOutputCommiter() method. This output commiter is invoked by Hadoop, so nothing has to be included in the Map or Reduce setup/cleanup methods. We are broken this contract in MOUTs as we are calling the OutputCommiter in the cleanup() method.

2. MOUTs commit folder is the temporary folder used by the default OutputFormat. That means that MOUTs will move the results to the default OutputFormat temporary folder. That delegates the responsibility in moving the temporary folder the final destination to the OutputCommiter provided by the default Job OutputFormat.

In other words, commit is done in two steps. In the first step, MOUT data is moved to the default temporary folder when the cleanup() methods is called. In the second step, the temporary folder is moved to the final destination. This final step is done by the OutputCommiter returned by the getOutputCommiter() of the default OutputFormat.

Impact

• At the end, MOUT data will be commited to the final destination * only if the default OutputFormat inherits from FileOutputFormat*. In other cases, data will be keep in the _temporary folder (Not sure if it will be clean up or not). That happens because the data moving to the final destination is only implemented in the OutputCommiter of the FileOutputFormat class.
• Two task attemps that success can collide when committing to the temporary directory using MOUT. That happens because the call to the commit is done in the cleanup() method, that can be called in both task attemps. Instead the commit should have been invoked by Hadoop over just one task attemp. It is probable that this can explain some of the issues we find when running some jobs.
• Proper methods of the MOUT OutputCommiter will not be called when a job is killed or aborted. I'm specifically talking about methods like OutputCommiter.abortTask() or OutputCommiter.abortJob(). I don't know the impact of that.

Solution

The solution should be capture the signals in the the default ProxyOutputCommiter and distribute them to MOUT's particular OutputCommiters.

[pereferrera]

Just to add to this, I want to remember that the only main OutputFormat in a Pangool Job is a ProxyOutputFormat which inherits from FileOutputFormat, so the first point in "Impact" is not relevant.

I will analyze the other two.

[ivanprado]

Yes, it is relevant. Although you are right, that the default OutputFormat
of a Pangool Job is a ProxyOutputFormat, it is also true that the commiting
responsibility is delegated entirely to the underlying OutputFormat. So
Both points are relevant.

It is also true that because we already have a ProxyOutputFormat by
default, it can be used to capture the OutputCommiter signals and send them
to the MOUTs OutputCommiters.

2013/10/1 Pere Ferrera notifications@github.com

Just to add to this, I want to remember that the only main OutputFormat in
a Pangool Job is a ProxyOutputFormat which inherits from FileOutputFormat,
so the first point in "Impact" is not relevant.

I will analyze the other two.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/32#issuecomment-25442683
.

Iván de Prado
CEO & Co-founder
www.datasalt.com

[pereferrera]

I still don't get the point with the main OutputFormat.

The only OutputFormat possible for a Pangool Job is ProxyOutputFormat. (Don't be confused with the "main" OutputFormat that the user can specify, which is the one being proxied. This one is non-relevant to this discussion.)

So, the committing is delegated to the ProxyOutputFormat. We already programmed it thinking about that (and that's why it extends FileOutputFormat). So I don't see the problem in your first point. Maybe you can reformulate it with different sentences and terms in case I didn't get something relevant.

However I can understand the problems related with speculative execution, but let me get it clear:

"Two task attemps that success can collide when committing to the temporary directory using MOUT. That happens because the call to the commit is done in the cleanup() method, that can be called in both task attemps. Instead the commit should have been invoked by Hadoop over just one task attempt."

Maybe you mean that two tasks attempts run speculatively, one succeeds and the other one is killed, but both are called cleanup(). Is that what you meant?

But let's talk about a possible solution. Because the current API is just mimicking Hadoop's solution for this. In Hadoop, MultipleOutputs are not a special OutputFormat, but also some kind of wrapper that is managed in a Mapper or a Reducer. I can imagine a possible solution would be:

• To create a special OutputFormat. It would receive three things when writing: A name and a pair of (Key, Value) of any kind.
• This OutputFormat would delegate to any RecordWriter and manage the lifecycle properly.
• A Pangool Job would always use this OutputFormat. If the Job is configured only with one Output, the API should be adapted so that writing to the normal Collector redirects to the only output of this new OutputFormat.

It would involve changes in everything related to MultipleOutputs (a major refactor), in the TupleMRBuilder, in the MapOnlyJobBuilder, in all Mappers / Reducers classes and in the Collector classes.

It doesn't seem very complex. But it involves a lot of refactoring.

Is the proposed solution correct or do you have something else in mind?

[pereferrera]

Let's also think about the fact that now it's possible to write to the main (user) OutputFormat and have a folder structure with the "extra" outputs. If we create a NamedOutputsOutputFormat, how should we handle this from now on?

Or should we still have a ProxyOutputFormat that uses both a (user-defined main OutputFormat) and maybe a collection of extra OutputFormats organized in folders (in the case of FileOutputFormats) like now?

[ivanprado]

I see what you say about the ProxyOutputCommiter. Looking carefully I see
that is taking care of the commit in this line:

super.commitTask(taskContext);

So, my first point was irrelevant as you told. I will think more in a
solution, although I'm not sure if it is worth because involves a lot of
changes for not so much goodness.

2013/10/1 Pere Ferrera notifications@github.com

Let's also think about the fact that now it's possible to write to the
main (user) OutputFormat and have a folder structure with the "extra"
outputs. If we create a NamedOutputsOutputFormat, how should we handle this
from now on?

Or should we still have a ProxyOutputFormat that uses both a (user-defined
main OutputFormat) and maybe a collection of extra OutputFormats organized
in folders (in the case of FileOutputFormats) like now?

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/32#issuecomment-25444373
.

Iván de Prado
CEO & Co-founder
www.datasalt.com