[
https://issues.apache.org/jira/browse/MAPREDUCE-6166?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14237016#comment-14237016
]
Gera Shegalov commented on MAPREDUCE-6166:
------------------------------------------
Thanks, It makes sense, [~eepayne]. I just wanted to confirm that there are no
existing tests catching missing checksum in on-disk shuffle ([confirmed!|by
https://issues.apache.org/jira/browse/MAPREDUCE-6166?focusedCommentId=14236700&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14236700]).
For the production code I have only one comment:
I am now convinced {{FileSystem.getLocal(conf)}} should be
{{FileSystem.getLocal(conf).getRaw()}}
In the corresponding {{OnDiskMapOutput}} constructor.
{{TestFetcher#testCorruptedFiles}} comments:
Nit: Lower-case {{FETCHER}} because it's a final variable.
{{FileSystem fs}} need to be changed {{FileSystem.getLocal(conf).getRaw()}}}
just like in production code, and it could be made an instance variable because
we can use it for cleanup in {{tearDown}}.
{{Path p}} should be more mnemonic, and it it's better to use a root directory
that matches the test method so we can use it for cleanup in {{tearDown}}
{code}
Path outputPath = new Path(name.getMethodName() + "/foo");
{code}
Instead of reverse-engineering the path {{shuffledToDisk}}, we can use
{code}Path shuffledToDisk = OnDiskMapOutput.getTempPath(outputPath,
fetcher);{code}
{quote}
{code}
457 ios.write(mapData.getBytes());
458 ios.close();
{code}
{quote}
{{ios.close()}} should be in a finally block.
{quote}
{code}
476 bin = new ByteArrayInputStream(corrupted);
477 // Read past the shuffle header.
478 bin.read(new byte[headerSize], 0, headerSize);
{code}
{quote}
Move lines 477-478 inside the following try on 480.
Drop {{fs.deleteOnExit}}. It comes too late in case there was an exception
before, we should better do cleanup inside the {{tearDown}} method.
{quote}
{code}
491 IFileInputStream iFin = new IFileInputStream(
492 new FileInputStream(shuffledToDisk.toString()), dataSize, job);
{code}
{quote}
It's probably better not to mix in java.io API if we already use Hadoop
FileSystem API. Why not do:
{code}
491 IFileInputStream iFin = new
IFileInputStream(fs.open(shuffledToDisk), dataSize, job);
{code}
{quote}
{code}
493 iFin.read(new byte[dataSize], 0, dataSize);
494 iFin.close();
495 fs.close();
{code}
{quote}
Make sure to put 493, 494 in try/finally, accordingly.
Since we are getting rid of {{fs.deleteOnExit}} we don't need {{fs.close}}
> Reducers do not catch bad map output transfers during shuffle if data
> shuffled directly to disk
> -----------------------------------------------------------------------------------------------
>
> Key: MAPREDUCE-6166
> URL: https://issues.apache.org/jira/browse/MAPREDUCE-6166
> Project: Hadoop Map/Reduce
> Issue Type: Bug
> Components: mrv2
> Affects Versions: 2.6.0
> Reporter: Eric Payne
> Assignee: Eric Payne
> Attachments: MAPREDUCE-6166-gera-missing-cs-test.patch,
> MAPREDUCE-6166.v1.201411221941.txt, MAPREDUCE-6166.v2.201411251627.txt,
> MAPREDUCE-6166.v3.txt
>
>
> In very large map/reduce jobs (50000 maps, 2500 reducers), the intermediate
> map partition output gets corrupted on disk on the map side. If this
> corrupted map output is too large to shuffle in memory, the reducer streams
> it to disk without validating the checksum. In jobs this large, it could take
> hours before the reducer finally tries to read the corrupted file and fails.
> Since retries of the failed reduce attempt will also take hours, this delay
> in discovering the failure is multiplied greatly.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
