MapReduce Spill Failure

(16 posts) (3 voices)

• Started 1 year ago by
• Latest reply from
• This topic is resolved

Tags:

• hadoop
• mapreduce

1. 

   Anonymous
   Unregistered

   I am having trouble running my map reduce job on my DSE 2.0-1 cluster. When I run the job, the mappers run for a while on all the analytics nodes in my cluster. When they try to spill the mapper output to disk, though, they fail. The message in the log is:
   Caused by: org.apache.hadoop.util.DiskChecker$DiskErrorException: Could not find any valid local directory for output/spill1.out

   I have tried setting mapred.local.dir in the $DSE_HOME/resources/hadoop/conf/taskcontroller.cfg file but I get the same error.

   Any advice would be appreciated.

   Posted 1 year ago #
2. 

   tjake
   Moderator

   Sounds like you need to setup your mapred.local.dir on the nodes.

   <property>
   <name>mapred.local.dir</name>
   <value>/d1/mapred/local,/d2/mapred/local,/d3/mapred/local,/d4/mapred/local</value>
   </property>

   This defaults to /tmp/hadoop-${user.name}

   Posted 1 year ago #
3. 

   Anonymous
   Unregistered

   I set that property in both $DSE_HOME/resources/hadoop/conf/taskcontroller.cfg and added the xml property declaration to $DSE_HOME/resources/hadoop/conf/dse-core.xml but I still get the same error.

   Posted 1 year ago #
4. 

   tjake
   Moderator

   Can you post the job.xml file generated for your job?

   Posted 1 year ago #
5. 

   Anonymous
   Unregistered

   Here's the job.xml:

   <?xml version="1.0" encoding="UTF-8" standalone="no"?><configuration>
   <property><name>fs.s3n.impl</name><value>org.apache.hadoop.fs.s3native.NativeS3FileSystem</value></property>
   <property><name>mapred.task.cache.levels</name><value>2</value></property>
   <property><name>mapred.job.inode.mode</name><value>SYNC</value></property>
   <property><name>hadoop.tmp.dir</name><value>/tmp/hadoop-${user.name}</value></property>
   <property><name>hadoop.native.lib</name><value>true</value></property>
   <property><name>map.sort.class</name><value>org.apache.hadoop.util.QuickSort</value></property>
   <property><name>dse.cfs.repair.batchsize</name><value>300</value></property>
   <property><name>ipc.client.idlethreshold</name><value>4000</value></property>
   <property><name>cassandra.thrift.address</name><value>nyarchapp01d</value></property>
   <property><name>mapred.system.dir</name><value>${hadoop.tmp.dir}/mapred/system</value></property>
   <property><name>mapred.job.tracker.persist.jobstatus.hours</name><value>0</value></property>
   <property><name>io.skip.checksum.errors</name><value>false</value></property>
   <property><name>fs.default.name</name><value>cfs:///</value></property>
   <property><name>mapred.cluster.reduce.memory.mb</name><value>-1</value></property>
   <property><name>mapred.reducer.new-api</name><value>true</value></property>
   <property><name>mapred.child.tmp</name><value>./tmp</value></property>
   <property><name>fs.har.impl.disable.cache</name><value>true</value></property>
   <property><name>cassandra.partitioner.class</name><value>org.apache.cassandra.dht.RandomPartitioner</value></property>
   <property><name>mapred.skip.reduce.max.skip.groups</name><value>0</value></property>
   <property><name>mapred.heartbeats.in.second</name><value>100</value></property>
   <property><name>mapred.tasktracker.dns.nameserver</name><value>default</value></property>
   <property><name>io.sort.factor</name><value>10</value></property>
   <property><name>cassandra.input.columnfamily</name><value>MyEvent</value></property>
   <property><name>mapred.task.timeout</name><value>600000</value></property>
   <property><name>mapred.max.tracker.failures</name><value>4</value></property>
   <property><name>hadoop.rpc.socket.factory.class.default</name><value>org.apache.hadoop.net.StandardSocketFactory</value></property>
   <property><name>mapred.job.tracker.jobhistory.lru.cache.size</name><value>5</value></property>
   <property><name>fs.hdfs.impl</name><value>org.apache.hadoop.hdfs.DistributedFileSystem</value></property>
   <property><name>mapred.queue.default.acl-administer-jobs</name><value>*</value></property>
   <property><name>mapred.output.key.class</name><value>org.apache.hadoop.io.Text</value></property>
   <property><name>mapred.skip.map.auto.incr.proc.count</name><value>true</value></property>
   <property><name>mapreduce.job.complete.cancel.delegation.tokens</name><value>true</value></property>
   <property><name>io.mapfile.bloom.size</name><value>1048576</value></property>
   <property><name>mapreduce.reduce.shuffle.connect.timeout</name><value>180000</value></property>
   <property><name>mapred.jobtracker.blacklist.fault-timeout-window</name><value>180</value></property>
   <property><name>tasktracker.http.threads</name><value>40</value></property>
   <property><name>mapred.job.shuffle.merge.percent</name><value>0.66</value></property>
   <property><name>mapreduce.inputformat.class</name><value>org.apache.cassandra.hadoop.ColumnFamilyInputFormat</value></property>
   <property><name>fs.ftp.impl</name><value>org.apache.hadoop.fs.ftp.FTPFileSystem</value></property>
   <property><name>mapred.output.compress</name><value>false</value></property>
   <property><name>io.bytes.per.checksum</name><value>512</value></property>
   <property><name>mapred.combine.recordsBeforeProgress</name><value>10000</value></property>
   <property><name>mapred.healthChecker.script.timeout</name><value>600000</value></property>
   <property><name>topology.node.switch.mapping.impl</name><value>org.apache.hadoop.net.ScriptBasedMapping</value></property>
   <property><name>cassandra.thrift.port</name><value>9160</value></property>
   <property><name>mapred.reduce.slowstart.completed.maps</name><value>0.05</value></property>
   <property><name>mapred.reduce.max.attempts</name><value>4</value></property>
   <property><name>cassandra.input.keyspace</name><value>ReportingKeyspace</value></property>
   <property><name>fs.ramfs.impl</name><value>org.apache.hadoop.fs.InMemoryFileSystem</value></property>
   <property><name>mapred.skip.map.max.skip.records</name><value>0</value></property>
   <property><name>mapred.cluster.map.memory.mb</name><value>-1</value></property>
   <property><name>fs.local.block.size</name><value>67108864</value></property>
   <property><name>hadoop.security.group.mapping</name><value>org.apache.hadoop.security.ShellBasedUnixGroupsMapping</value></property>
   <property><name>mapred.job.tracker.persist.jobstatus.dir</name><value>/jobtracker/jobsInfo</value></property>
   <property><name>mapred.jar</name><value>cfs:/tmp/hadoop-apparch/mapred/staging/apparch/.staging/job_201203301707_0004/job.jar</value></property>
   <property><name>fs.s3.buffer.dir</name><value>${hadoop.tmp.dir}/s3</value></property>
   <property><name>job.end.retry.attempts</name><value>0</value></property>
   <property><name>dse.consistencylevel.read</name><value>ONE</value></property>
   <property><name>fs.file.impl</name><value>org.apache.hadoop.fs.LocalFileSystem</value></property>
   <property><name>mapred.local.dir.minspacestart</name><value>0</value></property>
   <property><name>mapred.output.compression.type</name><value>RECORD</value></property>
   <property><name>cassandra.input.predicate</name><value>0f00010b000000060000000c6f66666572696e67546b65790000000a646576696365436f64650000000d616666696c69617465436f64650000000b73657276696365546b65790000001373746172744576656e7454696d657374616d7000000015656e64456c6170736564506c61795365636f6e647300</value></property>
   <property><name>dfs.permissions</name><value>false</value></property>
   <property><name>topology.script.number.args</name><value>100</value></property>
   <property><name>io.mapfile.bloom.error.rate</name><value>0.005</value></property>
   <property><name>mapred.cluster.max.reduce.memory.mb</name><value>-1</value></property>
   <property><name>mapred.max.tracker.blacklists</name><value>4</value></property>
   <property><name>mapred.task.profile.maps</name><value>0-2</value></property>
   <property><name>mapred.userlog.retain.hours</name><value>24</value></property>
   <property><name>mapred.job.tracker.persist.jobstatus.active</name><value>false</value></property>
   <property><name>hadoop.security.authorization</name><value>false</value></property>
   <property><name>local.cache.size</name><value>10737418240</value></property>
   <property><name>mapred.min.split.size</name><value>0</value></property>
   <property><name>mapred.map.tasks</name><value>3</value></property>
   <property><name>fs.cfs.impl</name><value>com.datastax.bdp.hadoop.cfs.CassandraFileSystem</value></property>
   <property><name>mapred.child.java.opts</name><value>-Xmx256m</value></property>
   <property><name>mapreduce.job.counters.limit</name><value>120</value></property>
   <property><name>mapred.output.value.class</name><value>org.apache.hadoop.io.SortedMapWritable</value></property>
   <property><name>dse.autoconfigure.node</name><value>false</value></property>
   <property><name>mapred.job.queue.name</name><value>default</value></property>
   <property><name>mapred.job.tracker.retiredjobs.cache.size</name><value>1000</value></property>
   <property><name>ipc.server.listen.queue.size</name><value>128</value></property>
   <property><name>job.end.retry.interval</name><value>30000</value></property>
   <property><name>mapred.inmem.merge.threshold</name><value>1000</value></property>
   <property><name>mapred.skip.attempts.to.start.skipping</name><value>2</value></property>
   <property><name>mapreduce.tasktracker.outofband.heartbeat.damper</name><value>1000000</value></property>
   <property><name>fs.checkpoint.dir</name><value>${hadoop.tmp.dir}/dfs/namesecondary</value></property>
   <property><name>mapred.reduce.tasks</name><value>1</value></property>
   <property><name>mapred.merge.recordsBeforeProgress</name><value>10000</value></property>
   <property><name>mapred.userlog.limit.kb</name><value>0</value></property>
   <property><name>mapred.job.reduce.memory.mb</name><value>-1</value></property>
   <property><name>webinterface.private.actions</name><value>false</value></property>
   <property><name>hadoop.security.token.service.use_ip</name><value>true</value></property>
   <property><name>io.sort.spill.percent</name><value>0.80</value></property>
   <property><name>mapred.job.shuffle.input.buffer.percent</name><value>0.70</value></property>
   <property><name>mapred.job.name</name><value>offering count</value></property>
   <property><name>mapred.map.tasks.speculative.execution</name><value>true</value></property>
   <property><name>hadoop.util.hash.type</name><value>murmur</value></property>
   <property><name>mapred.map.max.attempts</name><value>4</value></property>
   <property><name>mapreduce.job.acl-view-job</name><value> </value></property>
   <property><name>mapred.job.tracker.handler.count</name><value>10</value></property>
   <property><name>mapreduce.reduce.shuffle.read.timeout</name><value>180000</value></property>
   <property><name>mapred.tasktracker.expiry.interval</name><value>600000</value></property>
   <property><name>mapred.jobtracker.maxtasks.per.job</name><value>-1</value></property>
   <property><name>mapred.jobtracker.job.history.block.size</name><value>3145728</value></property>
   <property><name>keep.failed.task.files</name><value>false</value></property>
   <property><name>mapred.task.profile.reduces</name><value>0-2</value></property>
   <property><name>ipc.client.tcpnodelay</name><value>false</value></property>
   <property><name>mapred.output.compression.codec</name><value>org.apache.hadoop.io.compress.DefaultCodec</value></property>
   <property><name>io.map.index.skip</name><value>0</value></property>
   <property><name>mapred.working.dir</name><value>cfs:/user/apparch</value></property>
   <property><name>ipc.server.tcpnodelay</name><value>false</value></property>
   <property><name>mapred.jobtracker.blacklist.fault-bucket-width</name><value>15</value></property>
   <property><name>mapred.used.genericoptionsparser</name><value>true</value></property>
   <property><name>mapred.mapper.new-api</name><value>true</value></property>
   <property><name>mapred.job.map.memory.mb</name><value>-1</value></property>
   <property><name>hadoop.logfile.size</name><value>10000000</value></property>
   <property><name>mapred.reduce.tasks.speculative.execution</name><value>true</value></property>
   <property><name>mapreduce.job.dir</name><value>cfs:/tmp/hadoop-apparch/mapred/staging/apparch/.staging/job_201203301707_0004</value></property>
   <property><name>mapreduce.tasktracker.outofband.heartbeat</name><value>false</value></property>
   <property><name>mapreduce.reduce.input.limit</name><value>-1</value></property>
   <property><name>hadoop.security.authentication</name><value>simple</value></property>
   <property><name>fs.checkpoint.period</name><value>3600</value></property>
   <property><name>mapred.job.reuse.jvm.num.tasks</name><value>24</value></property>
   <property><name>mapred.jobtracker.completeuserjobs.maximum</name><value>100</value></property>
   <property><name>jobtracker.thrift.address</name><value>0.0.0.0:9290</value></property>
   <property><name>mapred.task.tracker.task-controller</name><value>org.apache.hadoop.mapred.DefaultTaskController</value></property>
   <property><name>fs.s3.maxRetries</name><value>4</value></property>
   <property><name>mapred.cluster.max.map.memory.mb</name><value>-1</value></property>
   <property><name>mapreduce.reduce.shuffle.maxfetchfailures</name><value>10</value></property>
   <property><name>mapreduce.job.acl-modify-job</name><value> </value></property>
   <property><name>mapred.local.dir</name><value>${hadoop.tmp.dir}/mapred/local</value></property>
   <property><name>fs.hftp.impl</name><value>org.apache.hadoop.hdfs.HftpFileSystem</value></property>
   <property><name>fs.trash.interval</name><value>0</value></property>
   <property><name>fs.s3.sleepTimeSeconds</name><value>10</value></property>
   <property><name>mapred.submit.replication</name><value>10</value></property>
   <property><name>fs.har.impl</name><value>org.apache.hadoop.fs.HarFileSystem</value></property>
   <property><name>mapred.map.output.compression.codec</name><value>com.datastax.bdp.hadoop.compression.SnappyCodec</value></property>
   <property><name>mapred.tasktracker.dns.interface</name><value>default</value></property>
   <property><name>mapred.job.tracker</name><value>${dse.job.tracker}</value></property>
   <property><name>dse.consistencylevel.write</name><value>ONE</value></property>
   <property><name>mapreduce.job.submithost</name><value>nyarchapp01d</value></property>
   <property><name>io.seqfile.sorter.recordlimit</name><value>1000000</value></property>
   <property><name>mapred.line.input.format.linespermap</name><value>1</value></property>
   <property><name>mapred.jobtracker.taskScheduler</name><value>org.apache.hadoop.mapred.JobQueueTaskScheduler</value></property>
   <property><name>fs.webhdfs.impl</name><value>org.apache.hadoop.hdfs.web.WebHdfsFileSystem</value></property>
   <property><name>mapred.local.dir.minspacekill</name><value>0</value></property>
   <property><name>io.sort.record.percent</name><value>0.05</value></property>
   <property><name>mapreduce.reduce.class</name><value>com.myorg.hadoop.MyEventReducer</value></property>
   <property><name>fs.kfs.impl</name><value>org.apache.hadoop.fs.kfs.KosmosFileSystem</value></property>
   <property><name>mapred.temp.dir</name><value>${hadoop.tmp.dir}/mapred/temp</value></property>
   <property><name>mapred.tasktracker.reduce.tasks.maximum</name><value>2</value></property>
   <property><name>fs.checkpoint.edits.dir</name><value>${fs.checkpoint.dir}</value></property>
   <property><name>mapred.tasktracker.tasks.sleeptime-before-sigkill</name><value>5000</value></property>
   <property><name>mapred.job.reduce.input.buffer.percent</name><value>0.0</value></property>
   <property><name>mapred.tasktracker.indexcache.mb</name><value>10</value></property>
   <property><name>mapred.jobtracker.plugins</name><value>com.datastax.bdp.hadoop.thriftfs.ThriftJobTrackerPlugin</value></property>
   <property><name>mapreduce.job.split.metainfo.maxsize</name><value>10000000</value></property>
   <property><name>mapred.skip.reduce.auto.incr.proc.count</name><value>true</value></property>
   <property><name>hadoop.logfile.count</name><value>10</value></property>
   <property><name>mapreduce.job.submithostaddress</name><value>10.40.199.130</value></property>
   <property><name>io.seqfile.compress.blocksize</name><value>1048576</value></property>
   <property><name>fs.s3.block.size</name><value>67108864</value></property>
   <property><name>mapred.tasktracker.taskmemorymanager.monitoring-interval</name><value>5000</value></property>
   <property><name>mapred.queue.default.state</name><value>RUNNING</value></property>
   <property><name>mapred.acls.enabled</name><value>false</value></property>
   <property><name>mapreduce.jobtracker.staging.root.dir</name><value>${hadoop.tmp.dir}/mapred/staging</value></property>
   <property><name>mapred.queue.names</name><value>default</value></property>
   <property><name>fs.hsftp.impl</name><value>org.apache.hadoop.hdfs.HsftpFileSystem</value></property>
   <property><name>mapred.task.tracker.http.address</name><value>0.0.0.0:50060</value></property>
   <property><name>mapreduce.combine.class</name><value>com.myorg.hadoop.MyEventReducer</value></property>
   <property><name>fs.cfs-archive.impl</name><value>com.datastax.bdp.hadoop.cfs.CassandraFileSystem</value></property>
   <property><name>mapred.reduce.parallel.copies</name><value>5</value></property>
   <property><name>io.seqfile.lazydecompress</name><value>true</value></property>
   <property><name>mapred.output.dir</name><value>/tmp/hadoopout</value></property>
   <property><name>io.sort.mb</name><value>100</value></property>
   <property><name>ipc.client.connection.maxidletime</name><value>10000</value></property>
   <property><name>mapred.compress.map.output</name><value>true</value></property>
   <property><name>hadoop.security.uid.cache.secs</name><value>14400</value></property>
   <property><name>mapred.task.tracker.report.address</name><value>127.0.0.1:0</value></property>
   <property><name>mapred.healthChecker.interval</name><value>60000</value></property>
   <property><name>ipc.client.kill.max</name><value>10</value></property>
   <property><name>ipc.client.connect.max.retries</name><value>10</value></property>
   <property><name>mapreduce.map.class</name><value>com.myorg.hadoop.MyEventMapper</value></property>
   <property><name>fs.s3.impl</name><value>org.apache.hadoop.fs.s3.S3FileSystem</value></property>
   <property><name>mapred.user.jobconf.limit</name><value>5242880</value></property>
   <property><name>mapred.job.tracker.http.address</name><value>0.0.0.0:50030</value></property>
   <property><name>io.file.buffer.size</name><value>4096</value></property>
   <property><name>mapred.jobtracker.restart.recover</name><value>false</value></property>
   <property><name>io.serializations</name><value>org.apache.hadoop.io.serializer.WritableSerialization</value></property>
   <property><name>mapred.task.profile</name><value>false</value></property>
   <property><name>jobclient.output.filter</name><value>FAILED</value></property>
   <property><name>mapred.tasktracker.map.tasks.maximum</name><value>2</value></property>
   <property><name>io.compression.codecs</name><value>org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.BZip2Codec</value></property>
   <property><name>fs.checkpoint.size</name><value>67108864</value></property>
   <property><name>fs.local.subblock.size</name><value>2097152</value></property>
   </configuration>

   Posted 1 year ago #
6. 

   tjake
   Moderator

   According to this your mapred.local.dir is ${hadoop.tmp.dir}/mapred/local

   Is your /tmp dir small?

   Can you try setting this to fixed directory in mapred-site.xml ?

   Posted 1 year ago #
7. 

   Anonymous
   Unregistered

   I didn't think it was too small (it had 6 GB free and my column family on which I was running the job was only 600 MB). When I changed the directory to an explicit dir on a filesystem with more space, it worked. Apparently, my mapper has a bug and it's generating way too much output.

   Thanks for the help.

   Posted 1 year ago #
8. 

   Stephanie
   Administrator

   Hi Yossarian,

   Were you able to get help with everything?

   Posted 1 year ago #
9. 

   Anonymous
   Unregistered

   I thought I did, but upon re-working my Mapper, I'm still having issues. I don't understand why my mapper is spilling at all... my column family has a bunch of columns but in the slice predicate, I'm just looking at 2 of them. My mapper is quite simple: for each row, it looks at column1 and if the value is > Threshold, it writes the value of column2 and an IntWritable set to 1 to the context.

   My reducer just sums up the ints and outputs the total for each key.

   Even though my test column family only contains 300 MB of data, approximately 10 GB of spill files are produced. I can see in the spill files that it contains not only the value of column2, but column1 as well (which I never write to the context). I thought the spill files should only contain the output of the map operation that doesn't fit in memory any more?

   Posted 1 year ago #
10. 

    tjake
    Moderator

    Can you avoid range slicing? do you know the columns you need? if so you can set cassandra.slice.predicate.column_names

    Posted 1 year ago #
11. 

    Anonymous
    Unregistered

    As far as I can tell, I am setting that... in my Job configuration, I do the following:

    List<ByteBuffer> colList = new ArrayList<ByteBuffer>();
      colList.add(ByteBuffer.wrap("column1".getBytes("UTF-8")));
      colList.add(ByteBuffer.wrap("column2".getBytes("UTF-8")));
      SlicePredicate predicate = new SlicePredicate().setColumn_names(colList);
      //snip
      ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);

    Posted 1 year ago #
12. 

    tjake
    Moderator

    Yeah that's fine. I have no idea why 10G of spill is generated from 600M of data. What exactly is the mapper doing with the data?

    Posted 1 year ago #
13. 

    Anonymous
    Unregistered

    Barely anything. Here's the map method in its entirety:

    public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) throws IOException, InterruptedException {
            IColumn col1 = (IColumn)columns.get(ByteBuffer.wrap("column1".getBytes("UTF-8")));
            IColumn col2 =(IColumn) columns.get(ByteBuffer.wrap("column2".getBytes("UTF-8")));
            if(col1 != null && col1.value()!=null && col2 != null && col2.value()!=null ){
                long value = ByteBufferUtil.toLong(col1.value());
                if(value > 0 && value <= THRESHOLD){
                    Text outputKey =      null;
                    try{
                        outputKey =   new Text(col2.value().array());
                    }catch(Exception e){
                        System.out.println("key is null");
                    }
                    if(outputKey != null){
                        context.write(outputKey,new IntWritable(1));
                    }
                }
            }
    
        }

    Posted 1 year ago #
14. 

    tjake
    Moderator

    col2.value().array() isn't safe (could generate a huge amount of garbage. use ByteBufferUtil.string(col2.value()), try that

    Posted 1 year ago #
15. 

    Anonymous
    Unregistered

    That solved my spill problem. One last question and I'll be all set:

    My mapper runs fine now and the job succeeds. The output file, however, looks like it didn't aggregate the results (all counts are 1 and the same key shows up multiple times in the output file). My reducer and job configuration look exactly like the WordCount example as far as I can tell (included below). Can you see anything I'm doing wrong?

    public static class MyReducer extends Reducer<Text, Iterable<IntWritable>, Text, IntWritable> {
            private IntWritable result = new IntWritable();
    
            public void reduce(Text key, Iterable<IntWritable> counts, Context context) throws Exception {
                int sum = 0;
                for (IntWritable count : counts) {
                    sum += count.get();
                }
                result.set(sum);
                context.write(key, result);
    
            }
        }

    And in the job configuration, I have:
    

    job.setCombinerClass(MyReducer.class);
     job.setReducerClass(MyReducer.class);

    Posted 1 year ago #

RSS feed for this topic

Reply »

You must log in to post.