2) Run it on fully-dis= tributed cluster: it always works fine with 20 data files input in fully-di= stributed cluster, while It always fails on pseudo-distributed cluster.

So, it seems to be related to the=A0xceiver/load limits= you mentioned, and I have changed the xceiver value in the hdfs-site.xml:<= /div>


<property>
        <name>dfs.datanode.max.xcievers</name>
        <value>4096</value>
</property>

but=A0I still got the same= error when running with 20 data files inputs in pseudo-distributed cluster= s.

How could I fix this problem?

2012/9/18 Harsh J &= lt;harsh@cloudera.c= om>

Hi Jason,

How many unique keys are you going to be generating from this program, roug= hly?

By default, the max-load of a DN is about 4k threads and if you're
trying to push beyond that value then the NN will no longer select the
DN as it would consider it already overloaded. In a fully distributed
mode, you may not see this issue as there's several DNs and TTs to
distribute the write load across.

Try with a smaller input sample if there's a whole lot of keys you'= ll
be creating files for, and see if that works instead (such that
there's fewer files and you do not hit the xceiver/load limits).

On Mon, Sep 17, 2012 at 7:20 PM, Jason Yang <lin.yang.jason@gmail.com> wrote:
> Hi, all
>
> I have written a simple MR program which partition a file into multipl= e
> files bases on the clustering result of the points in this file, here = is my
> code:
> ---
> private int run() throws IOException
> {
> String scheme =3D getConf().get(CommonUtility.ATTR_SCHEME);
> String ecgDir =3D getConf().get(CommonUtility.ATTR_ECG_DATA_DIR);
> String outputDir =3D getConf().get(CommonUtility.ATTR_OUTPUT_DIR);
>
> // create JobConf
> JobConf jobConf =3D new JobConf(getConf(), this.getClass());
>
> // set path for input and output
> Path inPath =3D new Path(scheme + ecgDir);
> Path outPath =3D new Path(scheme + outputDir +
> CommonUtility.OUTPUT_LOCAL_CLUSTERING);
> FileInputFormat.addInputPath(jobConf, inPath);
> FileOutputFormat.setOutputPath(jobConf, outPath);
>
> // clear output if it already existed
> CommonUtility.deleteHDFSFile(outPath.toString());
>
> // set format for input and output
> jobConf.setInputFormat(WholeFileInputFormat.class);
> jobConf.setOutputFormat(LocalClusterMSFOutputFormat.class);
>
> // set class of output key and value
> jobConf.setOutputKeyClass(Text.class);
> jobConf.setOutputValueClass(RRIntervalWritable.class);
>
> // set mapper and reducer
> jobConf.setMapperClass(LocalClusteringMapper.class);
> jobConf.setReducerClass(IdentityReducer.class);
>
>
> // run the job
> JobClient.runJob(jobConf);
> return 0;
> }
>
> ...
>
> public class LocalClusteringMapper extends MapReduceBase implements > Mapper<NullWritable, BytesWritable, Text, RRIntervalWritable> > {
> @Override
> public void map(NullWritable key, BytesWritable value,
> OutputCollector<Text, RRIntervalWritable> output, Reporter repor= ter)
> throws IOException
> {
> //read and cluster
> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 ...
>
> // output
> Iterator<RRIntervalWritable> it =3D rrArray.iterator();
> while (it.hasNext())
> {
> RRIntervalWritable rr =3D it.next();
>
> Text outputKey =3D new Text(rr.clusterResult );
>
> output.collect(outputKey, rr);
> }
>
> }
>
> ...
>
> public class LocalClusterMSFOutputFormat extends
> MultipleSequenceFileOutputFormat<Text, RRIntervalWritable>
> {
>
> protected String generateFileNameForKeyValue(Text key,
> RRIntervalWritable value, String name)
> {
> return value.clusterResult.toString();
> }
> }
> ---
>
> But this program always get a IO Exception when running in a
> pseudo-distributed cluster, and the log has been attached at the end o= f this
> post.
>
> There's something wired:
> 1. If I use the SequenceFileOutputFormat instead of
> MultipleSequenceFileOutputFormat, this program would works fine( at le= ast
> there is no error in log).
> 2. The one which always cause the error is the EcgData002509_LCF_3
>
>
>>
>> 12/09/17 21:10:35 INFO mapred.MapTask: Starting flush of map outpu= t
>> 12/09/17 21:10:35 INFO mapred.MapTask: Finished spill 0
>> 12/09/17 21:10:35 INFO mapred.TaskRunner:
>> Task:attempt_local_0001_m_000019_0 is done. And is in the process = of
>> commiting
>> 12/09/17 21:10:35 INFO mapred.LocalJobRunner:
>> 12/09/17 21:10:35 INFO mapred.TaskRunner: Task
>> 'attempt_local_0001_m_000019_0' done.
>> 12/09/17 21:10:35 INFO mapred.LocalJobRunner:
>> 12/09/17 21:10:35 INFO mapred.Merger: Merging 20 sorted segments >> 12/09/17 21:10:35 INFO mapred.Merger: Merging 2 intermediate segme= nts out
>> of a total of 20
>> 12/09/17 21:10:35 INFO mapred.Merger: Merging 10 intermediate segm= ents out
>> of a total of 19
>> 12/09/17 21:10:35 INFO mapred.Merger: Down to the last merge-pass,= with 10
>> segments left of total size: 18913891 bytes
>> 12/09/17 21:10:35 INFO mapred.LocalJobRunner:
>> 12/09/17 21:10:39 WARN hdfs.DFSClient: DataStreamer Exception:
>> org.apache.hadoop.ipc.RemoteException: java.io.IOException: File >> /work/lab/output/localClustering/_temporary/_attempt_local_0001_r_= 000000_0/EcgData002509_LCF_3
>> could only be replicated to 0 nodes, instead of 1
>> at
>> org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalB= lock(FSNamesystem.java:1271)
>> at
>> org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.= java:422)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:396)
>> at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
>> at org.apache.hadoop.ipc.Client.call(Client.java:740)
>> at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
>> at $Proxy0.addBlock(Unknown Source)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(Ret= ryInvocationHandler.java:82)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvo= cationHandler.java:59)
>> at $Proxy0.addBlock(Unknown Source)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.locateFollowingBl= ock(DFSClient.java:2937)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.nextBlockOutputSt= ream(DFSClient.java:2819)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.access$2000(DFSCl= ient.java:2102)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream$DataStreamer.run(= DFSClient.java:2288)
>> 12/09/17 21:10:39 WARN hdfs.DFSClient: Error Recovery for block nu= ll bad
>> datanode[0] nodes =3D=3D null
>> 12/09/17 21:10:39 WARN hdfs.DFSClient: Could not get block locatio= ns.
>> Source file
>> "/work/lab/output/localClustering/_temporary/_attempt_local_0= 001_r_000000_0/EcgData002509_LCF_3"
>> - Aborting...
>> 12/09/17 21:10:41 INFO mapred.LocalJobRunner: reduce > reduce >> 12/09/17 21:10:42 INFO mapred.JobClient: =A0map 100% reduce 89% >> 12/09/17 21:10:42 WARN mapred.LocalJobRunner: job_local_0001
>> org.apache.hadoop.ipc.RemoteException: java.io.IOException: File >> /work/lab/output/localClustering/_temporary/_attempt_local_0001_r_= 000000_0/EcgData002509_LCF_3
>> could only be replicated to 0 nodes, instead of 1
>> at
>> org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalB= lock(FSNamesystem.java:1271)
>> at
>> org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.= java:422)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:396)
>> at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
>> at org.apache.hadoop.ipc.Client.call(Client.java:740)
>> at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
>> at $Proxy0.addBlock(Unknown Source)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(Ret= ryInvocationHandler.java:82)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvo= cationHandler.java:59)
>> at $Proxy0.addBlock(Unknown Source)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.locateFollowingBl= ock(DFSClient.java:2937)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.nextBlockOutputSt= ream(DFSClient.java:2819)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.access$2000(DFSCl= ient.java:2102)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream$DataStreamer.run(= DFSClient.java:2288)
>> 12/09/17 21:10:43 INFO mapred.JobClient: Job complete: job_local_0= 001
>> 12/09/17 21:10:43 INFO mapred.JobClient: Counters: 15
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 FileSystemCounters >> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 FILE_BYTES_READ= =3D23297226
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 HDFS_BYTES_READ= =3D546711709
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 FILE_BYTES_WRITTE= N=3D232075142
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 HDFS_BYTES_WRITTE= N=3D13530993
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 Map-Reduce Framework<= br> >> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Reduce input grou= ps=3D56
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Combine output re= cords=3D0
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Map input records= =3D20
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Reduce shuffle by= tes=3D0
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Reduce output rec= ords=3D38837
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Spilled Records= =3D102562
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Map output bytes= =3D18691072
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Map input bytes= =3D28649088
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Combine input rec= ords=3D0
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Map output record= s=3D55700
>> 12/09/17 21:10:43 INFO mapred.JobClient: =A0 =A0 Reduce input reco= rds=3D38838
>> 12/09/17 21:10:44 INFO mapred.LocalJobRunner: reduce > reduce >> Exception in thread "main" java.io.IOException: Job fail= ed!
>> at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1252)<= br> >> at org.yanglin.mr.lab.ecg.PESCJob.runLocalClustering(PESCJob.java:= 111)
>> at org.yanglin.mr.lab.ecg.PESCJob.run(PESCJob.java:57)
>> at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
>> at org.yanglin.mr.lab.ecg.PESCJob.main(PESCJob.java:117)
>> 12/09/17 21:10:48 ERROR hdfs.DFSClient: Exception closing file
>> /work/lab/output/localClustering/_temporary/_attempt_local_0001_r_= 000000_0/EcgData002509_LCF_3
>> : org.apache.hadoop.ipc.RemoteException: java.io.IOException: File=
>> /work/lab/output/localClustering/_temporary/_attempt_local_0001_r_= 000000_0/EcgData002509_LCF_3
>> could only be replicated to 0 nodes, instead of 1
>> at
>> org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalB= lock(FSNamesystem.java:1271)
>> at
>> org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.= java:422)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:396)
>> at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
>> org.apache.hadoop.ipc.RemoteException: java.io.IOException: File >> /work/lab/output/localClustering/_temporary/_attempt_local_0001_r_= 000000_0/EcgData002509_LCF_3
>> could only be replicated to 0 nodes, instead of 1
>> at
>> org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalB= lock(FSNamesystem.java:1271)
>> at
>> org.apache.hadoop.hdfs.server.namenode.NameNode.addBlock(NameNode.= java:422)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:508)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:959)
>> at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:955)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:396)
>> at org.apache.hadoop.ipc.Server$Handler.run(Server.java:953)
>> at org.apache.hadoop.ipc.Client.call(Client.java:740)
>> at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:220)
>> at $Proxy0.addBlock(Unknown Source)
>> at sun.reflect.GeneratedMethodAccessor7.invoke(Unknown Source)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAc= cessorImpl.java:25)
>> at java.lang.reflect.Method.invoke(Method.java:597)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(Ret= ryInvocationHandler.java:82)
>> at
>> org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvo= cationHandler.java:59)
>> at $Proxy0.addBlock(Unknown Source)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.locateFollowingBl= ock(DFSClient.java:2937)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.nextBlockOutputSt= ream(DFSClient.java:2819)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.access$2000(DFSCl= ient.java:2102)
>> at
>> org.apache.hadoop.hdfs.DFSClient$DFSOutputStream$DataStreamer.run(= DFSClient.java:2288)
>
>
>
>
> --
> YANG, Lin
>

--
Harsh J

YANG, Lin

--20cf307cfd36094dc104c9f1b50d--