You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-user@hadoop.apache.org by tim robertson <ti...@gmail.com> on 2008/11/27 10:55:24 UTC
Newbie: Problem splitting a tab file into many (>20,000) files
Hi all,
I am really struggling with splitting a single file into many files
using hadoop and would appreciate any help offered. The input file is
150,000,000 rows long today, but will grow to 1Billion+.
My mapper simply emits a key that it determines from the data (key
will be used for the output file name), and then I have no reduction,
but a MultipleTextOutputFormat that provides the output file name.
This works for small input files just fine, but with a large input
file that will result in 20,000 or so output files (maximum possible
long term will be 64,000 files), it dies with various IOExceptions
depending on how the cluster is feeling at the time it seems.
Sometimes it dies after the mapping, sometimes during it with "too
many files open" problems. (I am running 20 small EC2 instances)
Is there a better approach to this problem? Any suggestions welcome!
Please bear in mind that as soon as I split them, I will use up to
1000 of those files as Input Paths for the next job. Will I hit
problems having 1000 inputs to a job?
Maybe I should use some Identity Reducers? Will lots of them help
hadoop handle so many files? I don't care about sorting, so have
reducers set to 0 currently.
Thanks for any help with this task!
Tim
The code:
public void map(LongWritable key, Text value, OutputCollector<Text,
Text> output, Reporter reporter) throws IOException {
String data = value.toString();
try {
String parts[] = tabPattern.split(data); // just RegEX based splitting
double y = Double.parseDouble(parts[10]);
double x = Double.parseDouble(parts[11]);
int cell = CellIdUtils.toCellId(new Double(y).floatValue(), new
Double(x).floatValue());
occurrence.set(data);
cellId.set("" + cell);
output.collect(cellId, occurrence); // key becomes file name - see below
} catch(Exception e) {}
}
public class CellBasedOutputFormat extends
MultipleTextOutputFormat<WritableComparable, Writable> {
@Override
protected String generateFileNameForKeyValue(WritableComparable
key,Writable value, String name) {
return "cell_" + key.toString();
}
}
RE: Hadoop Internal Architecture writeup (failure handling)
Posted by Amar Kamat <am...@yahoo-inc.com>.
@Ques-7 : mapred.jobtracker.restart.recover is the parameter that enables the feature for hadoop versions > 0.19. mapred.jobtracker.job.history.block.size controls the extent of recovery.
Amar
-----Original Message-----
From: Amar Kamat [mailto:amarrk@yahoo-inc.com]
Sent: Mon 12/1/2008 12:25 PM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup (failure handling)
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Mon 12/1/2008 10:09 AM
To: core-user@hadoop.apache.org
Subject: Hadoop Internal Architecture writeup (failure handling)
More questions on component failure handling. Can anyone confirm (or correct) that ?
1) When a TaskTracker crashes, the JobTracker haven't heard its heartbeat after a timeout period will conclude its crashes and re-allocate the unfinished task to other tasktrackers. Correct ?
All the tasks (complete + incomplete) from running jobs with their output on local filesystem will be re-executed. Only incomplete tasks from running jobs with their output on HDFS will be re-executed. Its also okay to say that maps from running jobs with num-reducers > 0 will be re-executed (completed + incomplete).
2) If the original TaskTracker is just overloaded rather than crash, what if it submit the Task result afterwards ?
Always the task (i.e the attempt) that reports first will be considered and other attempts will be killed.
3) When a DataNode crashes, the NameNode haven't heard its heartbeat after a timeout period will conclude its crashes. The NameNode will gradually redistribute the data chunks of the failed DataNode to other DataNodes to comply with the replication factors. Correct ?
4) Now what if the crashed DataNode reboot after an hour and rejoin the cluster. It will report to the NameNode which chunks it has. How does the NameNode recognize which chunks are outdated ? Using some kind of version number ?
5) After NameNode detects certain chunks are outdated, does NameNode simply discard them (and garbage collect them later) or does it try to bring them up to date (e.g. send it all the deltas) ?
6) When a NameNode crashes, all HDFS write cannot proceed. But HDFS reads can proceed if the client has a chunk handle to the DataNode. Correct ? This situation will continue until the NameNode is recovered and things will be back to normal. Correct ?
7) When a JobTracker crashes, all the Job that hasn't been completed will all be discarded. When the JobTracker restarted, the client need to resubmit all the jobs again. Correct ?
Yes. This is true for hadoop versions < 0.19. With hadoop-0.19 we have a fault tolerant jobtracker. So all the submitted jobs will be recovered and running jobs will be resumed. The extent of recovery might not be 100%.
Rgds,
Ricky
-----Original Message-----
From: Amar Kamat [mailto:amarrk@yahoo-inc.com]
Sent: Sunday, November 30, 2008 8:02 PM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Hey, nice work and nice writeup. Keep it up.
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Fri 11/28/2008 9:45 AM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Amar, thanks a lot. This is exactly the kind of feedback that I am looking for ... I have some more question ...
==================
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
==================
[Ricky] What exactly does the job.split contains ? I assume it contains the specification for each split (but not its data), such as what is the corresponding file and the byte range within that file. Correct ?
====================
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
=====================
[Ricky] I am curious about why can't the reduce execution start earlier (before all the map tasks completed). The value "iterator" inside the used-defined reduce() method can be blocked to wait for more map tasks completion. In other words, the map() and reduce() can also be proceeding in a pipeline parallelism.
======================
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
======================
[Ricky] Do you mean if the job has 5000 splits, then it requires 5000 TaskTrackers VM (one for each split) ?
comment:
If the job has 5000 splits, then it requires 5000 VMs (one for each split). TaskTracker is a framework daemon. TaskTracker is a process (JVM) that handles/manages tasks (processes processing a split) on a node. A TaskTracker is recognized by (node-hostname + port). A task is never executed in a TaskTracker and new jvm is spawned. The reason being that a faulty usercode(map/reduce) should not bring down a TaskTracker (a framework process). But with hadoop-0.19 we have jvm reuse and hence 5000 splits might require < 5000 VMs. Note that tasks in the end might get speculated which might add to the VM count.
Amar
=======================
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
========================
[Ricky] Is this a well-know folder within the HDFS ?
This is set using "mapred.system.dir" during cluster startup (see hadoop-default.conf). Its a framework directory.
RE: Hadoop Internal Architecture writeup (failure handling)
Posted by Amar Kamat <am...@yahoo-inc.com>.
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Mon 12/1/2008 10:09 AM
To: core-user@hadoop.apache.org
Subject: Hadoop Internal Architecture writeup (failure handling)
More questions on component failure handling. Can anyone confirm (or correct) that ?
1) When a TaskTracker crashes, the JobTracker haven't heard its heartbeat after a timeout period will conclude its crashes and re-allocate the unfinished task to other tasktrackers. Correct ?
All the tasks (complete + incomplete) from running jobs with their output on local filesystem will be re-executed. Only incomplete tasks from running jobs with their output on HDFS will be re-executed. Its also okay to say that maps from running jobs with num-reducers > 0 will be re-executed (completed + incomplete).
2) If the original TaskTracker is just overloaded rather than crash, what if it submit the Task result afterwards ?
Always the task (i.e the attempt) that reports first will be considered and other attempts will be killed.
3) When a DataNode crashes, the NameNode haven't heard its heartbeat after a timeout period will conclude its crashes. The NameNode will gradually redistribute the data chunks of the failed DataNode to other DataNodes to comply with the replication factors. Correct ?
4) Now what if the crashed DataNode reboot after an hour and rejoin the cluster. It will report to the NameNode which chunks it has. How does the NameNode recognize which chunks are outdated ? Using some kind of version number ?
5) After NameNode detects certain chunks are outdated, does NameNode simply discard them (and garbage collect them later) or does it try to bring them up to date (e.g. send it all the deltas) ?
6) When a NameNode crashes, all HDFS write cannot proceed. But HDFS reads can proceed if the client has a chunk handle to the DataNode. Correct ? This situation will continue until the NameNode is recovered and things will be back to normal. Correct ?
7) When a JobTracker crashes, all the Job that hasn't been completed will all be discarded. When the JobTracker restarted, the client need to resubmit all the jobs again. Correct ?
Yes. This is true for hadoop versions < 0.19. With hadoop-0.19 we have a fault tolerant jobtracker. So all the submitted jobs will be recovered and running jobs will be resumed. The extent of recovery might not be 100%.
Rgds,
Ricky
-----Original Message-----
From: Amar Kamat [mailto:amarrk@yahoo-inc.com]
Sent: Sunday, November 30, 2008 8:02 PM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Hey, nice work and nice writeup. Keep it up.
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Fri 11/28/2008 9:45 AM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Amar, thanks a lot. This is exactly the kind of feedback that I am looking for ... I have some more question ...
==================
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
==================
[Ricky] What exactly does the job.split contains ? I assume it contains the specification for each split (but not its data), such as what is the corresponding file and the byte range within that file. Correct ?
====================
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
=====================
[Ricky] I am curious about why can't the reduce execution start earlier (before all the map tasks completed). The value "iterator" inside the used-defined reduce() method can be blocked to wait for more map tasks completion. In other words, the map() and reduce() can also be proceeding in a pipeline parallelism.
======================
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
======================
[Ricky] Do you mean if the job has 5000 splits, then it requires 5000 TaskTrackers VM (one for each split) ?
comment:
If the job has 5000 splits, then it requires 5000 VMs (one for each split). TaskTracker is a framework daemon. TaskTracker is a process (JVM) that handles/manages tasks (processes processing a split) on a node. A TaskTracker is recognized by (node-hostname + port). A task is never executed in a TaskTracker and new jvm is spawned. The reason being that a faulty usercode(map/reduce) should not bring down a TaskTracker (a framework process). But with hadoop-0.19 we have jvm reuse and hence 5000 splits might require < 5000 VMs. Note that tasks in the end might get speculated which might add to the VM count.
Amar
=======================
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
========================
[Ricky] Is this a well-know folder within the HDFS ?
This is set using "mapred.system.dir" during cluster startup (see hadoop-default.conf). Its a framework directory.
Hadoop Internal Architecture writeup (failure handling)
Posted by Ricky Ho <rh...@adobe.com>.
More questions on component failure handling. Can anyone confirm (or correct) that ?
1) When a TaskTracker crashes, the JobTracker haven't heard its heartbeat after a timeout period will conclude its crashes and re-allocate the unfinished task to other tasktrackers. Correct ?
2) If the original TaskTracker is just overloaded rather than crash, what if it submit the Task result afterwards ?
3) When a DataNode crashes, the NameNode haven't heard its heartbeat after a timeout period will conclude its crashes. The NameNode will gradually redistribute the data chunks of the failed DataNode to other DataNodes to comply with the replication factors. Correct ?
4) Now what if the crashed DataNode reboot after an hour and rejoin the cluster. It will report to the NameNode which chunks it has. How does the NameNode recognize which chunks are outdated ? Using some kind of version number ?
5) After NameNode detects certain chunks are outdated, does NameNode simply discard them (and garbage collect them later) or does it try to bring them up to date (e.g. send it all the deltas) ?
6) When a NameNode crashes, all HDFS write cannot proceed. But HDFS reads can proceed if the client has a chunk handle to the DataNode. Correct ? This situation will continue until the NameNode is recovered and things will be back to normal. Correct ?
7) When a JobTracker crashes, all the Job that hasn't been completed will all be discarded. When the JobTracker restarted, the client need to resubmit all the jobs again. Correct ?
Rgds,
Ricky
-----Original Message-----
From: Amar Kamat [mailto:amarrk@yahoo-inc.com]
Sent: Sunday, November 30, 2008 8:02 PM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Hey, nice work and nice writeup. Keep it up.
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Fri 11/28/2008 9:45 AM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Amar, thanks a lot. This is exactly the kind of feedback that I am looking for ... I have some more question ...
==================
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
==================
[Ricky] What exactly does the job.split contains ? I assume it contains the specification for each split (but not its data), such as what is the corresponding file and the byte range within that file. Correct ?
====================
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
=====================
[Ricky] I am curious about why can't the reduce execution start earlier (before all the map tasks completed). The value "iterator" inside the used-defined reduce() method can be blocked to wait for more map tasks completion. In other words, the map() and reduce() can also be proceeding in a pipeline parallelism.
======================
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
======================
[Ricky] Do you mean if the job has 5000 splits, then it requires 5000 TaskTrackers VM (one for each split) ?
comment:
If the job has 5000 splits, then it requires 5000 VMs (one for each split). TaskTracker is a framework daemon. TaskTracker is a process (JVM) that handles/manages tasks (processes processing a split) on a node. A TaskTracker is recognized by (node-hostname + port). A task is never executed in a TaskTracker and new jvm is spawned. The reason being that a faulty usercode(map/reduce) should not bring down a TaskTracker (a framework process). But with hadoop-0.19 we have jvm reuse and hence 5000 splits might require < 5000 VMs. Note that tasks in the end might get speculated which might add to the VM count.
Amar
=======================
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
========================
[Ricky] Is this a well-know folder within the HDFS ?
This is set using "mapred.system.dir" during cluster startup (see hadoop-default.conf). Its a framework directory.
Re: Hadoop Internal Architecture writeup
Posted by Aaron Kimball <aa...@cloudera.com>.
Good stuff -- please link to this resource from the Hadoop wiki (on the
"articles" page?)
- Aaron
On Sun, Nov 30, 2008 at 8:01 PM, Amar Kamat <am...@yahoo-inc.com> wrote:
> Hey, nice work and nice writeup. Keep it up.
> Comments inline.
> Amar
>
>
> -----Original Message-----
> From: Ricky Ho [mailto:rho@adobe.com]
> Sent: Fri 11/28/2008 9:45 AM
> To: core-user@hadoop.apache.org
> Subject: RE: Hadoop Internal Architecture writeup
>
> Amar, thanks a lot. This is exactly the kind of feedback that I am looking
> for ... I have some more question ...
>
> ==================
> The jobclient while submitting the job calculates the split using
> InputFormat which is specified by the user. Internally the InputFormat
> might make use of dfs-block size, user-hinted num-maps etc. The
> jobtracker is given 3 files
> - job.xml : job control parameters
> - job.split : the split file
> - job.jar : user map-reduce code
> ==================
> [Ricky] What exactly does the job.split contains ? I assume it contains
> the specification for each split (but not its data), such as what is the
> corresponding file and the byte range within that file. Correct ?
>
> ====================
> This process is interleaved/parallelized. As soon as a map is done, the
> JobTracker is notified. Once a tracker (with a reducer) asks for events,
> these new events are passed. Hence the map output pulling (Shuffle
> Phase) works in parallel with the Map Phase. Reduce Phase can start only
> once all the (resp) map outputs are copied and merged.
> =====================
> [Ricky] I am curious about why can't the reduce execution start earlier
> (before all the map tasks completed). The value "iterator" inside the
> used-defined reduce() method can be blocked to wait for more map tasks
> completion. In other words, the map() and reduce() can also be proceeding
> in a pipeline parallelism.
>
>
> ======================
> There is a 1-1 mapping between a split and a map task. Hence it will
> re-run the map on the corresponding split.
> ======================
> [Ricky] Do you mean if the job has 5000 splits, then it requires 5000
> TaskTrackers VM (one for each split) ?
>
> comment:
> If the job has 5000 splits, then it requires 5000 VMs (one for each split).
> TaskTracker is a framework daemon. TaskTracker is a process (JVM) that
> handles/manages tasks (processes processing a split) on a node. A
> TaskTracker is recognized by (node-hostname + port). A task is never
> executed in a TaskTracker and new jvm is spawned. The reason being that a
> faulty usercode(map/reduce) should not bring down a TaskTracker (a framework
> process). But with hadoop-0.19 we have jvm reuse and hence 5000 splits might
> require < 5000 VMs. Note that tasks in the end might get speculated which
> might add to the VM count.
> Amar
>
>
> =======================
> The client is unblocked once the job is submitted. The way it works is
> as follows :
> - jobclient requests the jobtracker for a unique job id
> - jobclient does some sanity checks to see if the output folder exists
> etc ...
> - jobclient uploads job files (xml, jar, split) onto a known location
> called System-Directory
> ========================
> [Ricky] Is this a well-know folder within the HDFS ?
>
> This is set using "mapred.system.dir" during cluster startup (see
> hadoop-default.conf). Its a framework directory.
>
>
>
RE: Hadoop Internal Architecture writeup
Posted by Amar Kamat <am...@yahoo-inc.com>.
Hey, nice work and nice writeup. Keep it up.
Comments inline.
Amar
-----Original Message-----
From: Ricky Ho [mailto:rho@adobe.com]
Sent: Fri 11/28/2008 9:45 AM
To: core-user@hadoop.apache.org
Subject: RE: Hadoop Internal Architecture writeup
Amar, thanks a lot. This is exactly the kind of feedback that I am looking for ... I have some more question ...
==================
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
==================
[Ricky] What exactly does the job.split contains ? I assume it contains the specification for each split (but not its data), such as what is the corresponding file and the byte range within that file. Correct ?
====================
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
=====================
[Ricky] I am curious about why can't the reduce execution start earlier (before all the map tasks completed). The value "iterator" inside the used-defined reduce() method can be blocked to wait for more map tasks completion. In other words, the map() and reduce() can also be proceeding in a pipeline parallelism.
======================
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
======================
[Ricky] Do you mean if the job has 5000 splits, then it requires 5000 TaskTrackers VM (one for each split) ?
comment:
If the job has 5000 splits, then it requires 5000 VMs (one for each split). TaskTracker is a framework daemon. TaskTracker is a process (JVM) that handles/manages tasks (processes processing a split) on a node. A TaskTracker is recognized by (node-hostname + port). A task is never executed in a TaskTracker and new jvm is spawned. The reason being that a faulty usercode(map/reduce) should not bring down a TaskTracker (a framework process). But with hadoop-0.19 we have jvm reuse and hence 5000 splits might require < 5000 VMs. Note that tasks in the end might get speculated which might add to the VM count.
Amar
=======================
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
========================
[Ricky] Is this a well-know folder within the HDFS ?
This is set using "mapred.system.dir" during cluster startup (see hadoop-default.conf). Its a framework directory.
RE: Hadoop Internal Architecture writeup
Posted by Ricky Ho <rh...@adobe.com>.
Does this mean within a TaskTracker, all reduce() method (even with different key) has to be called sequentially within a single thread ? If so, this contract also restrict the use of multiple threads (ie: a different thread for each key).
Isn't it true that in most case, the application doesn't care the order of keys ? Because if the application requires certain keys to arrive at the same node, then it will be difficult to improve parallelism by just adding more reducer nodes.
Or is this contract a configurable option ?
May be I am missing the rationale behind this contract ? Can you shed some light ?
Rgds, Ricky
> [Ricky] I am curious about why can't the reduce execution start
> earlier (before all the map tasks completed).
The contract is that each reduce is given the keys in sorted order.
The reduce can't start until it is sure it has the first key. That can
only happen after the maps are all finished.
Re: Hadoop Internal Architecture writeup
Posted by Owen O'Malley <om...@apache.org>.
On Nov 28, 2008, at 9:45 AM, Ricky Ho wrote:
> [Ricky] What exactly does the job.split contains ? I assume it
> contains the specification for each split (but not its data), such
> as what is the corresponding file and the byte range within that
> file. Correct ?
Yes
> [Ricky] I am curious about why can't the reduce execution start
> earlier (before all the map tasks completed).
The contract is that each reduce is given the keys in sorted order.
The reduce can't start until it is sure it has the first key. That can
only happen after the maps are all finished.
> [Ricky] Do you mean if the job has 5000 splits, then it requires
> 5000 TaskTrackers VM (one for each split) ?
In 0.19, you can enable the framework to re-use jvms between tasks in
the same job. Look at HADOOP-249.
> [Ricky] Is this a well-know folder within the HDFS ?
It is configured by the cluster admin. However, applications should
*not* depend on the contents or even visibility of that directory. It
will almost certainly become inaccessible to clients as part of
increasing security.
-- Owen
RE: Hadoop Internal Architecture writeup
Posted by Ricky Ho <rh...@adobe.com>.
Amar, thanks a lot. This is exactly the kind of feedback that I am looking for ... I have some more question ...
==================
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
==================
[Ricky] What exactly does the job.split contains ? I assume it contains the specification for each split (but not its data), such as what is the corresponding file and the byte range within that file. Correct ?
====================
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
=====================
[Ricky] I am curious about why can't the reduce execution start earlier (before all the map tasks completed). The value "iterator" inside the used-defined reduce() method can be blocked to wait for more map tasks completion. In other words, the map() and reduce() can also be proceeding in a pipeline parallelism.
======================
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
======================
[Ricky] Do you mean if the job has 5000 splits, then it requires 5000 TaskTrackers VM (one for each split) ?
=======================
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
========================
[Ricky] Is this a well-know folder within the HDFS ?
Re: Hadoop Internal Architecture writeup
Posted by tim robertson <ti...@gmail.com>.
Hi Ricky,
As a newcomer to MR and Hadoop I think what you are doing is a great
addition to the docs. One thing I would like to see in this overview
is how JVM's are spawned in the process - e.g. is it 1 JVM per node
per job, or per node per task etc. The reason being it has
implications about setting up JVM wide in-memory caches for the Maps /
Reducers to utilise (E.g. a big lookup hashtable which the Map will
use). I assume it is a JVM per task per node to allow the task
tracker to kill the process and restrict memory to a task, but I am no
sure.
Thanks again
Tim
On Fri, Nov 28, 2008 at 7:59 AM, Amar Kamat <am...@yahoo-inc.com> wrote:
> Ricky Ho wrote:
>>
>> I put together an article describing the internal architecture of Hadoop
>> (HDFS, MapRed). I'd love to get some feedback if you see anything
>> inaccurate or missing ...
>> http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html
>>
>
> Few comments on MR :
> 1) "The JobTracker will first determine the number of splits (each split is
> configurable, ~16-64MB) from the input path, and select some TaskTracker
> based on their network proximity to the data sources, then the JobTracker
> send the task requests to those selected TaskTrackers."
> The jobclient while submitting the job calculates the split using
> InputFormat which is specified by the user. Internally the InputFormat might
> make use of dfs-block size, user-hinted num-maps etc. The jobtracker is
> given 3 files
> - job.xml : job control parameters
> - job.split : the split file
> - job.jar : user map-reduce code
> Hence the JobTracker never actually *calculates* the split.
> Also the task-tracker asks for the task. Its a pull method where the tracker
> ask and the jobtracker schedules based on (node/rack/switch) locality.
> JobTracker never actually initiates scheduling.
>
> 2) For each record parsed by the "InputFormat"
> To be more precise "RecordReader" included in the "InputFormat".
>
> 3) A periodic wakeup process will sort the memory buffer into different
> reducer node by invoke the "combine" function.
> Need to cross check if its a wakeup process or a on-demand thread that is
> spawned once the buffer is nearly full. Btw the function that determined
> which key-val goes to which reducer is called "Partitioner". Combiner is
> just an optimization that does a local merge/reduction/aggregation of the
> data before sending it over the network.
>
> 4) When all the TaskTrackers are done, the JobTracker will notify the
> selected TaskTrackers for the reduce phase.
> This process is interleaved/parallelized. As soon as a map is done, the
> JobTracker is notified. Once a tracker (with a reducer) asks for events,
> these new events are passed. Hence the map output pulling (Shuffle Phase)
> works in parallel with the Map Phase. Reduce Phase can start only once all
> the (resp) map outputs are copied and merged.
>
> 5) The JobTracker keep tracks of the progress of each phases and
> periodically ping the TaskTracker for their health status.
> Its again a push rather than a pull. Trackers report their status instead of
> JobTracker asking them.
>
> 6) When any of the map phase TaskTracker crashes, the JobTracker will
> reassign the map task to a different TaskTracker node, which will rerun all
> the assigned splits.
> There is a 1-1 mapping between a split and a map task. Hence it will re-run
> the map on the corresponding split.
>
> 7) After both phase completes, the JobTracker will unblock the client
> program.
> The client is unblocked once the job is submitted. The way it works is as
> follows :
> - jobclient requests the jobtracker for a unique job id
> - jobclient does some sanity checks to see if the output folder exists etc
> ...
> - jobclient uploads job files (xml, jar, split) onto a known location called
> System-Directory
> - jobclient informs the jobtracker that the files are ready and the
> jobtracker returns the control.
>
> Its not necessary that the jobclient always gets blocked on job submission.
>
> Amar
>>
>> Rgds,
>> Ricky
>>
>
>
Re: Hadoop Internal Architecture writeup
Posted by Amar Kamat <am...@yahoo-inc.com>.
Ricky Ho wrote:
> I put together an article describing the internal architecture of Hadoop (HDFS, MapRed). I'd love to get some feedback if you see anything inaccurate or missing ...
> http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html
>
Few comments on MR :
1) "The JobTracker will first determine the number of splits (each split
is configurable, ~16-64MB) from the input path, and select some
TaskTracker based on their network proximity to the data sources, then
the JobTracker send the task requests to those selected TaskTrackers."
The jobclient while submitting the job calculates the split using
InputFormat which is specified by the user. Internally the InputFormat
might make use of dfs-block size, user-hinted num-maps etc. The
jobtracker is given 3 files
- job.xml : job control parameters
- job.split : the split file
- job.jar : user map-reduce code
Hence the JobTracker never actually *calculates* the split.
Also the task-tracker asks for the task. Its a pull method where the
tracker ask and the jobtracker schedules based on (node/rack/switch)
locality. JobTracker never actually initiates scheduling.
2) For each record parsed by the “InputFormat”
To be more precise "RecordReader" included in the "InputFormat".
3) A periodic wakeup process will sort the memory buffer into different
reducer node by invoke the “combine” function.
Need to cross check if its a wakeup process or a on-demand thread that
is spawned once the buffer is nearly full. Btw the function that
determined which key-val goes to which reducer is called "Partitioner".
Combiner is just an optimization that does a local
merge/reduction/aggregation of the data before sending it over the network.
4) When all the TaskTrackers are done, the JobTracker will notify the
selected TaskTrackers for the reduce phase.
This process is interleaved/parallelized. As soon as a map is done, the
JobTracker is notified. Once a tracker (with a reducer) asks for events,
these new events are passed. Hence the map output pulling (Shuffle
Phase) works in parallel with the Map Phase. Reduce Phase can start only
once all the (resp) map outputs are copied and merged.
5) The JobTracker keep tracks of the progress of each phases and
periodically ping the TaskTracker for their health status.
Its again a push rather than a pull. Trackers report their status
instead of JobTracker asking them.
6) When any of the map phase TaskTracker crashes, the JobTracker will
reassign the map task to a different TaskTracker node, which will rerun
all the assigned splits.
There is a 1-1 mapping between a split and a map task. Hence it will
re-run the map on the corresponding split.
7) After both phase completes, the JobTracker will unblock the client
program.
The client is unblocked once the job is submitted. The way it works is
as follows :
- jobclient requests the jobtracker for a unique job id
- jobclient does some sanity checks to see if the output folder exists
etc ...
- jobclient uploads job files (xml, jar, split) onto a known location
called System-Directory
- jobclient informs the jobtracker that the files are ready and the
jobtracker returns the control.
Its not necessary that the jobclient always gets blocked on job submission.
Amar
> Rgds,
> Ricky
>
Re: Hadoop Internal Architecture writeup
Posted by Sharad Agarwal <sh...@yahoo-inc.com>.
Just glanced into it. Haven't read in detail. One correction about
Secondary namenode - It is not a Hot Standby.
see http://wiki.apache.org/hadoop/FAQ#7
Ricky Ho wrote:
> I put together an article describing the internal architecture of Hadoop (HDFS, MapRed). I'd love to get some feedback if you see anything inaccurate or missing ...
> http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html
>
> Rgds,
> Ricky
>
Hadoop Internal Architecture writeup
Posted by Ricky Ho <rh...@adobe.com>.
I put together an article describing the internal architecture of Hadoop (HDFS, MapRed). I'd love to get some feedback if you see anything inaccurate or missing ...
http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html
Rgds,
Ricky
Re: Newbie: Problem splitting a tab file into many (>20,000) files
Posted by tim robertson <ti...@gmail.com>.
Thanks Brian - I really appreciate the help and insights!
I have now moved to 0.19.0 as you suggest. I was indeed running heavy
Map tasks but was running only 1 per node.
I am attempting to cross reference 150M records each with a geospatial
Point (latitude and longitude) with a separate list of 120,000
polygons... PostGIS is performing really badly with this, and our
150million looks to increase to Billions, so I am looking for a
scalable process - hence Hadoop.
Since you recommend not splitting into many files (previously I was
splitting the 150M into lots of smaller files based on area and then
only using these as inputs to a polygon cross reference to reduce the
numbers), I am now able to keep the polygons in memory with a
geospatial index, and then just run over the full input file of point
records in one pass. I am struggling to make it only run 1 JVM per
node in 0.19.0 though, despite having:
<property>
<name>mapred.tasktracker.map.tasks.maximum</name>
<value>1</value>
</property>
<property>
<name>mapred.tasktracker.reduce.tasks.maximum</name>
<value>1</value>
</property>
Am I missing some configuration to enforce this?
Thanks all for any advice - I am new to hadoop...
Tim
On Fri, Nov 28, 2008 at 6:10 PM, Brian Bockelman <bb...@cse.unl.edu> wrote:
> Hey Tim,
>
> 1) You have Ganglia enabled, and it's unfortunately broken in that release.
> :( Turn it off, and you'll have much better luck.
> 2) The "failed to report status for 603 seconds" indicates to me that you
> might be overloading an individual datanode. Look at the Ganglia interface
> for the node and see if it is perhaps way overloaded or swapping? I see you
> are using 1.4GB per map task: each small instance has 1.7GB. Are you
> running multiple map tasks per node and going into swap?
> 3) Similarly, are you overloading your namenode through memory exhaustion?
> I'm no expert, but I'd speculate some of those messages could be causing
> this.
>
> It seems that you have a memory-intensive map. Could I recommend upgrading
> to 0.19.0 to take advantage of the improved JVM usage? I believe this would
> reduce the memory usage of your application greatly.
>
> Additionally, can I recommend that you re-visit the requirement to have 20k
> files outputted per job? Remember Hadoop is heavily optimized for
> medium-to-large size files. Is it not an option to use SequenceFile to
> combine those into a much smaller number of multi-hundred-MB files? You can
> probably make your setup work eventually, but it'll be a bit like fighting
> the tide. Alternately, if you must have random-record access, try putting
> your results into HBase.
>
> Hope this helps!
>
> Brian
>
> On Nov 28, 2008, at 2:14 AM, tim robertson wrote:
>
>> I guess no one is doing this kind of thing on hadoop?
>>
>> Here is my latest error (run on20 Ec2 small nodes, 200 Maps, 20
>> Reducers, -Xmx1400M).
>>
>> Any help, ideas for trying, greatly appreciated!
>>
>> Cheers,
>>
>> Tim
>>
>> ...
>> 08/11/27 12:31:23 INFO mapred.JobClient: map 100% reduce 73%
>> 08/11/27 12:31:36 INFO mapred.JobClient: map 100% reduce 74%
>> 08/11/27 12:32:09 INFO mapred.JobClient: map 100% reduce 75%
>> 08/11/27 12:41:50 INFO mapred.JobClient: map 100% reduce 71%
>> 08/11/27 12:41:50 INFO mapred.JobClient: Task Id :
>> attempt_200811271043_0008_r_000006_0, Status : FAILED
>> java.io.IOException: Could not get block locations. Aborting...
>> at
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> at
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> at
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>>
>> Task attempt_200811271043_0008_r_000006_0 failed to report status for
>> 603 seconds. Killing!
>> attempt_200811271043_0008_r_000006_0: Exception in thread "Timer
>> thread for monitoring mapred" java.lang.NullPointerException
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.ganglia.GangliaContext.xdr_string(GangliaContext.java:195)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.ganglia.GangliaContext.emitMetric(GangliaContext.java:138)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.ganglia.GangliaContext.emitRecord(GangliaContext.java:123)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.spi.AbstractMetricsContext.emitRecords(AbstractMetricsContext.java:304)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.spi.AbstractMetricsContext.timerEvent(AbstractMetricsContext.java:290)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.spi.AbstractMetricsContext.access$000(AbstractMetricsContext.java:50)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.metrics.spi.AbstractMetricsContext$1.run(AbstractMetricsContext.java:249)
>> attempt_200811271043_0008_r_000006_0: at
>> java.util.TimerThread.mainLoop(Timer.java:512)
>> attempt_200811271043_0008_r_000006_0: at
>> java.util.TimerThread.run(Timer.java:462)
>> attempt_200811271043_0008_r_000006_0: Exception closing file
>>
>> /tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10013.deflate
>> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
>> get block locations. Aborting...
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>> attempt_200811271043_0008_r_000006_0: Exception closing file
>>
>> /tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10031.deflate
>> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
>> get block locations. Aborting...
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> attempt_200811271043_0008_r_000006_0: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>> attempt_200811271043_0008_r_000006_0: Exception closing file
>>
>> /tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10068.deflate
>> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
>> get block locations. Aborting...
>> ..
>> .. Repeated 1000's times
>> ..
>> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
>> get block locations. Aborting...
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>> attempt_200811271043_0008_r_000018_2: Exception closing file
>>
>> /tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000018_2/cell_10420.deflate
>> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
>> get block locations. Aborting...
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>> attempt_200811271043_0008_r_000018_2: Exception closing file
>>
>> /tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000018_2/cell_10439.deflate
>> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
>> get block locations. Aborting...
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
>> attempt_200811271043_0008_r_000018_2: at
>>
>> org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
>> 08/11/27 12:54:56 INFO mapred.JobClient: map 100% reduce 47%
>> 08/11/27 12:55:02 INFO mapred.JobClient: map 100% reduce 48%
>> 08/11/27 12:55:07 INFO mapred.JobClient: map 100% reduce 50%
>> 08/11/27 12:55:09 INFO mapred.JobClient: map 100% reduce 52%
>> 08/11/27 12:55:15 INFO mapred.JobClient: map 100% reduce 53%
>> java.io.IOException: Job failed!
>> at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1113)
>> at
>> com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.splitOccurrenceDataIntoCells(OccurrenceByPolygonIntersection.java:99)
>> at
>> com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.run(OccurrenceByPolygonIntersection.java:55)
>> at
>> com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.main(OccurrenceByPolygonIntersection.java:206)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at org.apache.hadoop.util.RunJar.main(RunJar.java:155)
>> at org.apache.hadoop.mapred.JobShell.run(JobShell.java:54)
>> at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
>> at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:79)
>> at org.apache.hadoop.mapred.JobShell.main(JobShell.java:68)
>>
>>
>> On Thu, Nov 27, 2008 at 10:55 AM, tim robertson
>> <ti...@gmail.com> wrote:
>>>
>>> Hi all,
>>>
>>> I am really struggling with splitting a single file into many files
>>> using hadoop and would appreciate any help offered. The input file is
>>> 150,000,000 rows long today, but will grow to 1Billion+.
>>>
>>> My mapper simply emits a key that it determines from the data (key
>>> will be used for the output file name), and then I have no reduction,
>>> but a MultipleTextOutputFormat that provides the output file name.
>>>
>>> This works for small input files just fine, but with a large input
>>> file that will result in 20,000 or so output files (maximum possible
>>> long term will be 64,000 files), it dies with various IOExceptions
>>> depending on how the cluster is feeling at the time it seems.
>>> Sometimes it dies after the mapping, sometimes during it with "too
>>> many files open" problems. (I am running 20 small EC2 instances)
>>>
>>> Is there a better approach to this problem? Any suggestions welcome!
>>> Please bear in mind that as soon as I split them, I will use up to
>>> 1000 of those files as Input Paths for the next job. Will I hit
>>> problems having 1000 inputs to a job?
>>>
>>> Maybe I should use some Identity Reducers? Will lots of them help
>>> hadoop handle so many files? I don't care about sorting, so have
>>> reducers set to 0 currently.
>>>
>>> Thanks for any help with this task!
>>>
>>> Tim
>>>
>>> The code:
>>>
>>> public void map(LongWritable key, Text value, OutputCollector<Text,
>>> Text> output, Reporter reporter) throws IOException {
>>> String data = value.toString();
>>> try {
>>> String parts[] = tabPattern.split(data); // just RegEX based splitting
>>> double y = Double.parseDouble(parts[10]);
>>> double x = Double.parseDouble(parts[11]);
>>> int cell = CellIdUtils.toCellId(new Double(y).floatValue(), new
>>> Double(x).floatValue());
>>> occurrence.set(data);
>>> cellId.set("" + cell);
>>> output.collect(cellId, occurrence); // key becomes file name - see
>>> below
>>> } catch(Exception e) {}
>>> }
>>>
>>> public class CellBasedOutputFormat extends
>>> MultipleTextOutputFormat<WritableComparable, Writable> {
>>> @Override
>>> protected String generateFileNameForKeyValue(WritableComparable
>>> key,Writable value, String name) {
>>> return "cell_" + key.toString();
>>> }
>>> }
>>>
>
>
Re: Newbie: Problem splitting a tab file into many (>20,000) files
Posted by Brian Bockelman <bb...@cse.unl.edu>.
Hey Tim,
1) You have Ganglia enabled, and it's unfortunately broken in that
release. :( Turn it off, and you'll have much better luck.
2) The "failed to report status for 603 seconds" indicates to me that
you might be overloading an individual datanode. Look at the Ganglia
interface for the node and see if it is perhaps way overloaded or
swapping? I see you are using 1.4GB per map task: each small instance
has 1.7GB. Are you running multiple map tasks per node and going into
swap?
3) Similarly, are you overloading your namenode through memory
exhaustion? I'm no expert, but I'd speculate some of those messages
could be causing this.
It seems that you have a memory-intensive map. Could I recommend
upgrading to 0.19.0 to take advantage of the improved JVM usage? I
believe this would reduce the memory usage of your application greatly.
Additionally, can I recommend that you re-visit the requirement to
have 20k files outputted per job? Remember Hadoop is heavily
optimized for medium-to-large size files. Is it not an option to use
SequenceFile to combine those into a much smaller number of multi-
hundred-MB files? You can probably make your setup work eventually,
but it'll be a bit like fighting the tide. Alternately, if you must
have random-record access, try putting your results into HBase.
Hope this helps!
Brian
On Nov 28, 2008, at 2:14 AM, tim robertson wrote:
> I guess no one is doing this kind of thing on hadoop?
>
> Here is my latest error (run on20 Ec2 small nodes, 200 Maps, 20
> Reducers, -Xmx1400M).
>
> Any help, ideas for trying, greatly appreciated!
>
> Cheers,
>
> Tim
>
> ...
> 08/11/27 12:31:23 INFO mapred.JobClient: map 100% reduce 73%
> 08/11/27 12:31:36 INFO mapred.JobClient: map 100% reduce 74%
> 08/11/27 12:32:09 INFO mapred.JobClient: map 100% reduce 75%
> 08/11/27 12:41:50 INFO mapred.JobClient: map 100% reduce 71%
> 08/11/27 12:41:50 INFO mapred.JobClient: Task Id :
> attempt_200811271043_0008_r_000006_0, Status : FAILED
> java.io.IOException: Could not get block locations. Aborting...
> at org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> at org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
>
> Task attempt_200811271043_0008_r_000006_0 failed to report status for
> 603 seconds. Killing!
> attempt_200811271043_0008_r_000006_0: Exception in thread "Timer
> thread for monitoring mapred" java.lang.NullPointerException
> attempt_200811271043_0008_r_000006_0: at
> org
> .apache
> .hadoop
> .metrics.ganglia.GangliaContext.xdr_string(GangliaContext.java:195)
> attempt_200811271043_0008_r_000006_0: at
> org
> .apache
> .hadoop
> .metrics.ganglia.GangliaContext.emitMetric(GangliaContext.java:138)
> attempt_200811271043_0008_r_000006_0: at
> org
> .apache
> .hadoop
> .metrics.ganglia.GangliaContext.emitRecord(GangliaContext.java:123)
> attempt_200811271043_0008_r_000006_0: at
> org
> .apache
> .hadoop
> .metrics
> .spi.AbstractMetricsContext.emitRecords(AbstractMetricsContext.java:
> 304)
> attempt_200811271043_0008_r_000006_0: at
> org
> .apache
> .hadoop
> .metrics
> .spi.AbstractMetricsContext.timerEvent(AbstractMetricsContext.java:
> 290)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.metrics.spi.AbstractMetricsContext.access
> $000(AbstractMetricsContext.java:50)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.metrics.spi.AbstractMetricsContext
> $1.run(AbstractMetricsContext.java:249)
> attempt_200811271043_0008_r_000006_0: at
> java.util.TimerThread.mainLoop(Timer.java:512)
> attempt_200811271043_0008_r_000006_0: at
> java.util.TimerThread.run(Timer.java:462)
> attempt_200811271043_0008_r_000006_0: Exception closing file
> /tmp/occurrence/wdpa/split-1d/_temporary/
> _attempt_200811271043_0008_r_000006_0/cell_10013.deflate
> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
> get block locations. Aborting...
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
> attempt_200811271043_0008_r_000006_0: Exception closing file
> /tmp/occurrence/wdpa/split-1d/_temporary/
> _attempt_200811271043_0008_r_000006_0/cell_10031.deflate
> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
> get block locations. Aborting...
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> attempt_200811271043_0008_r_000006_0: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
> attempt_200811271043_0008_r_000006_0: Exception closing file
> /tmp/occurrence/wdpa/split-1d/_temporary/
> _attempt_200811271043_0008_r_000006_0/cell_10068.deflate
> attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
> get block locations. Aborting...
> ..
> .. Repeated 1000's times
> ..
> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
> get block locations. Aborting...
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
> attempt_200811271043_0008_r_000018_2: Exception closing file
> /tmp/occurrence/wdpa/split-1d/_temporary/
> _attempt_200811271043_0008_r_000018_2/cell_10420.deflate
> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
> get block locations. Aborting...
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
> attempt_200811271043_0008_r_000018_2: Exception closing file
> /tmp/occurrence/wdpa/split-1d/_temporary/
> _attempt_200811271043_0008_r_000018_2/cell_10439.deflate
> attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
> get block locations. Aborting...
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient
> $DFSOutputStream.processDatanodeError(DFSClient.java:2143)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access
> $1400(DFSClient.java:1735)
> attempt_200811271043_0008_r_000018_2: at
> org.apache.hadoop.dfs.DFSClient$DFSOutputStream
> $DataStreamer.run(DFSClient.java:1889)
> 08/11/27 12:54:56 INFO mapred.JobClient: map 100% reduce 47%
> 08/11/27 12:55:02 INFO mapred.JobClient: map 100% reduce 48%
> 08/11/27 12:55:07 INFO mapred.JobClient: map 100% reduce 50%
> 08/11/27 12:55:09 INFO mapred.JobClient: map 100% reduce 52%
> 08/11/27 12:55:15 INFO mapred.JobClient: map 100% reduce 53%
> java.io.IOException: Job failed!
> at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:
> 1113)
> at
> com
> .ibiodiversity
> .index
> .mapreduce
> .occurrence
> .geometry
> .OccurrenceByPolygonIntersection
> .splitOccurrenceDataIntoCells(OccurrenceByPolygonIntersection.java:99)
> at
> com
> .ibiodiversity
> .index
> .mapreduce
> .occurrence
> .geometry
> .OccurrenceByPolygonIntersection
> .run(OccurrenceByPolygonIntersection.java:55)
> at
> com
> .ibiodiversity
> .index
> .mapreduce
> .occurrence
> .geometry
> .OccurrenceByPolygonIntersection
> .main(OccurrenceByPolygonIntersection.java:206)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun
> .reflect
> .NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
> at
> sun
> .reflect
> .DelegatingMethodAccessorImpl
> .invoke(DelegatingMethodAccessorImpl.java:25)
> at java.lang.reflect.Method.invoke(Method.java:597)
> at org.apache.hadoop.util.RunJar.main(RunJar.java:155)
> at org.apache.hadoop.mapred.JobShell.run(JobShell.java:54)
> at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
> at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:79)
> at org.apache.hadoop.mapred.JobShell.main(JobShell.java:68)
>
>
> On Thu, Nov 27, 2008 at 10:55 AM, tim robertson
> <ti...@gmail.com> wrote:
>> Hi all,
>>
>> I am really struggling with splitting a single file into many files
>> using hadoop and would appreciate any help offered. The input file
>> is
>> 150,000,000 rows long today, but will grow to 1Billion+.
>>
>> My mapper simply emits a key that it determines from the data (key
>> will be used for the output file name), and then I have no reduction,
>> but a MultipleTextOutputFormat that provides the output file name.
>>
>> This works for small input files just fine, but with a large input
>> file that will result in 20,000 or so output files (maximum possible
>> long term will be 64,000 files), it dies with various IOExceptions
>> depending on how the cluster is feeling at the time it seems.
>> Sometimes it dies after the mapping, sometimes during it with "too
>> many files open" problems. (I am running 20 small EC2 instances)
>>
>> Is there a better approach to this problem? Any suggestions welcome!
>> Please bear in mind that as soon as I split them, I will use up to
>> 1000 of those files as Input Paths for the next job. Will I hit
>> problems having 1000 inputs to a job?
>>
>> Maybe I should use some Identity Reducers? Will lots of them help
>> hadoop handle so many files? I don't care about sorting, so have
>> reducers set to 0 currently.
>>
>> Thanks for any help with this task!
>>
>> Tim
>>
>> The code:
>>
>> public void map(LongWritable key, Text value, OutputCollector<Text,
>> Text> output, Reporter reporter) throws IOException {
>> String data = value.toString();
>> try {
>> String parts[] = tabPattern.split(data); // just RegEX based
>> splitting
>> double y = Double.parseDouble(parts[10]);
>> double x = Double.parseDouble(parts[11]);
>> int cell = CellIdUtils.toCellId(new Double(y).floatValue(), new
>> Double(x).floatValue());
>> occurrence.set(data);
>> cellId.set("" + cell);
>> output.collect(cellId, occurrence); // key becomes file name -
>> see below
>> } catch(Exception e) {}
>> }
>>
>> public class CellBasedOutputFormat extends
>> MultipleTextOutputFormat<WritableComparable, Writable> {
>> @Override
>> protected String generateFileNameForKeyValue(WritableComparable
>> key,Writable value, String name) {
>> return "cell_" + key.toString();
>> }
>> }
>>
Re: Newbie: Problem splitting a tab file into many (>20,000) files
Posted by tim robertson <ti...@gmail.com>.
I guess no one is doing this kind of thing on hadoop?
Here is my latest error (run on20 Ec2 small nodes, 200 Maps, 20
Reducers, -Xmx1400M).
Any help, ideas for trying, greatly appreciated!
Cheers,
Tim
...
08/11/27 12:31:23 INFO mapred.JobClient: map 100% reduce 73%
08/11/27 12:31:36 INFO mapred.JobClient: map 100% reduce 74%
08/11/27 12:32:09 INFO mapred.JobClient: map 100% reduce 75%
08/11/27 12:41:50 INFO mapred.JobClient: map 100% reduce 71%
08/11/27 12:41:50 INFO mapred.JobClient: Task Id :
attempt_200811271043_0008_r_000006_0, Status : FAILED
java.io.IOException: Could not get block locations. Aborting...
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
Task attempt_200811271043_0008_r_000006_0 failed to report status for
603 seconds. Killing!
attempt_200811271043_0008_r_000006_0: Exception in thread "Timer
thread for monitoring mapred" java.lang.NullPointerException
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.ganglia.GangliaContext.xdr_string(GangliaContext.java:195)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.ganglia.GangliaContext.emitMetric(GangliaContext.java:138)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.ganglia.GangliaContext.emitRecord(GangliaContext.java:123)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.spi.AbstractMetricsContext.emitRecords(AbstractMetricsContext.java:304)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.spi.AbstractMetricsContext.timerEvent(AbstractMetricsContext.java:290)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.spi.AbstractMetricsContext.access$000(AbstractMetricsContext.java:50)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.metrics.spi.AbstractMetricsContext$1.run(AbstractMetricsContext.java:249)
attempt_200811271043_0008_r_000006_0: at
java.util.TimerThread.mainLoop(Timer.java:512)
attempt_200811271043_0008_r_000006_0: at
java.util.TimerThread.run(Timer.java:462)
attempt_200811271043_0008_r_000006_0: Exception closing file
/tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10013.deflate
attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
get block locations. Aborting...
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
attempt_200811271043_0008_r_000006_0: Exception closing file
/tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10031.deflate
attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
get block locations. Aborting...
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
attempt_200811271043_0008_r_000006_0: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
attempt_200811271043_0008_r_000006_0: Exception closing file
/tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000006_0/cell_10068.deflate
attempt_200811271043_0008_r_000006_0: java.io.IOException: Could not
get block locations. Aborting...
..
.. Repeated 1000's times
..
attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
get block locations. Aborting...
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
attempt_200811271043_0008_r_000018_2: Exception closing file
/tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000018_2/cell_10420.deflate
attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
get block locations. Aborting...
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
attempt_200811271043_0008_r_000018_2: Exception closing file
/tmp/occurrence/wdpa/split-1d/_temporary/_attempt_200811271043_0008_r_000018_2/cell_10439.deflate
attempt_200811271043_0008_r_000018_2: java.io.IOException: Could not
get block locations. Aborting...
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
attempt_200811271043_0008_r_000018_2: at
org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
08/11/27 12:54:56 INFO mapred.JobClient: map 100% reduce 47%
08/11/27 12:55:02 INFO mapred.JobClient: map 100% reduce 48%
08/11/27 12:55:07 INFO mapred.JobClient: map 100% reduce 50%
08/11/27 12:55:09 INFO mapred.JobClient: map 100% reduce 52%
08/11/27 12:55:15 INFO mapred.JobClient: map 100% reduce 53%
java.io.IOException: Job failed!
at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1113)
at com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.splitOccurrenceDataIntoCells(OccurrenceByPolygonIntersection.java:99)
at com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.run(OccurrenceByPolygonIntersection.java:55)
at com.ibiodiversity.index.mapreduce.occurrence.geometry.OccurrenceByPolygonIntersection.main(OccurrenceByPolygonIntersection.java:206)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.apache.hadoop.util.RunJar.main(RunJar.java:155)
at org.apache.hadoop.mapred.JobShell.run(JobShell.java:54)
at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:79)
at org.apache.hadoop.mapred.JobShell.main(JobShell.java:68)
On Thu, Nov 27, 2008 at 10:55 AM, tim robertson
<ti...@gmail.com> wrote:
> Hi all,
>
> I am really struggling with splitting a single file into many files
> using hadoop and would appreciate any help offered. The input file is
> 150,000,000 rows long today, but will grow to 1Billion+.
>
> My mapper simply emits a key that it determines from the data (key
> will be used for the output file name), and then I have no reduction,
> but a MultipleTextOutputFormat that provides the output file name.
>
> This works for small input files just fine, but with a large input
> file that will result in 20,000 or so output files (maximum possible
> long term will be 64,000 files), it dies with various IOExceptions
> depending on how the cluster is feeling at the time it seems.
> Sometimes it dies after the mapping, sometimes during it with "too
> many files open" problems. (I am running 20 small EC2 instances)
>
> Is there a better approach to this problem? Any suggestions welcome!
> Please bear in mind that as soon as I split them, I will use up to
> 1000 of those files as Input Paths for the next job. Will I hit
> problems having 1000 inputs to a job?
>
> Maybe I should use some Identity Reducers? Will lots of them help
> hadoop handle so many files? I don't care about sorting, so have
> reducers set to 0 currently.
>
> Thanks for any help with this task!
>
> Tim
>
> The code:
>
> public void map(LongWritable key, Text value, OutputCollector<Text,
> Text> output, Reporter reporter) throws IOException {
> String data = value.toString();
> try {
> String parts[] = tabPattern.split(data); // just RegEX based splitting
> double y = Double.parseDouble(parts[10]);
> double x = Double.parseDouble(parts[11]);
> int cell = CellIdUtils.toCellId(new Double(y).floatValue(), new
> Double(x).floatValue());
> occurrence.set(data);
> cellId.set("" + cell);
> output.collect(cellId, occurrence); // key becomes file name - see below
> } catch(Exception e) {}
> }
>
> public class CellBasedOutputFormat extends
> MultipleTextOutputFormat<WritableComparable, Writable> {
> @Override
> protected String generateFileNameForKeyValue(WritableComparable
> key,Writable value, String name) {
> return "cell_" + key.toString();
> }
> }
>