Strange TempHfs error - wip-1.1

[Nate Murray]

Greetings,
I am getting a very strange TempHfs error, I was wondering if you
could take a look.

My flow looks like the following:

importPipe = new Each(importPipe, f, new Identity(minimalFields));

// group by request_id
importPipe = new GroupBy(importPipe, new Fields("request_id"));
importPipe = new Every(importPipe, Fields.ALL, new First(),
Fields.RESULTS);

// count the terms
importPipe = new GroupBy(importPipe, new Fields("search_term"));
importPipe = new Every(importPipe, new Count(new Fields
("term_count")));

// create a frequency
importPipe = new GroupBy(importPipe, new Fields("term_count")); //
<---- fails with this line

Flow parsedLogFlow = flowConnector.connect(sourceTap, sinkTap,
importPipe);
// etc

The stack trace is posted below.

A couple things are strange about this

1) If remove the GroupBy "term_count", every thing works fine. A file
is written to the HDFS (or Lfs) with the search_term and the count.
2) If I look up the MapReduce job in my browser the map and reduce
jobs are all zero. (That is to say, 0 pending, 0 running, 0 complete,
0 killed, 0 failed/killed task attempts). This means I don't get a
stack trace anywhere as to what the actual error is.

I simply get "step 2/3" failed, which appears to be saving to the
TempHfs.

Any thoughts on

1) how to get the full stack-trace of the actual error (I don't mind
editing the cascading code and rebuilding the jar
2) what the problem might be?

btw, I am using wip-1.1 branch from github

Thanks so much,

Nate


[exec] 09/10/20 08:45:11 INFO wordcount.Main:
toolkit.columnIndices: 7,8,0,11
[exec] 09/10/20 08:45:11 INFO wordcount.Main: toolkit.dateFormat:
yyyy-MM-dd HH:mm:ss
[exec] 09/10/20 08:45:11 INFO wordcount.Main:
toolkit.columnSeparator:
[exec] 09/10/20 08:45:11 INFO wordcount.Main:
mapred.reduce.tasks: 1
[exec] 09/10/20 08:45:11 INFO wordcount.Main:
toolkit.pageNumberIndex: null
[exec] 09/10/20 08:45:11 INFO wordcount.Main:
toolkit.pageIdTypeIndex: null
[exec] 09/10/20 08:45:12 INFO flow.MultiMapReducePlanner: using
application jar: /Users/xxx/projects/hadoop-jobs/query-hist/./build/
wordcount.jar
[exec] 09/10/20 08:45:12 INFO cascade.Cascade: Concurrent, Inc -
Cascading 1.1.0-${build.number} [hadoop-0.19.0+]
[exec] 09/10/20 08:45:12 INFO flow.Flow: [import] starting
[exec] 09/10/20 08:45:12 INFO flow.Flow: [import] source: Lfs
["TextLine[['offset', 'line']->[ALL]]"]["data/searches.seq"]"]
[exec] 09/10/20 08:45:12 INFO flow.Flow: [import] sink: Lfs
["TextLine[['offset', 'line']->[ALL]]"]["output"]"]
[exec] 09/10/20 08:45:12 INFO tap.Hfs: forcing job to local mode,
via source: Lfs["TextLine[['offset', 'line']->[ALL]]"]["data/
searches.seq"]"]
[exec] 09/10/20 08:45:13 INFO tap.Hfs: forcing job to local mode,
via sink: Lfs["TextLine[['offset', 'line']->[ALL]]"]["output"]"]
[exec] 09/10/20 08:45:13 INFO flow.Flow: [import] parallel
execution is enabled: true
[exec] 09/10/20 08:45:13 INFO flow.Flow: [import] starting jobs:
3
[exec] 09/10/20 08:45:13 INFO flow.Flow: [import] allocating
threads: 3
[exec] 09/10/20 08:45:13 INFO flow.FlowStep: [import] starting
step: (1/3) TempHfs["SequenceFile[['timestamp', 'visitor_id',
'search_term', 'request_id', 'ts']]"][import/42309/]
[exec] 09/10/20 08:45:13 INFO jvm.JvmMetrics: Initializing JVM
Metrics with processName=JobTracker, sessionId=
[exec] 09/10/20 08:45:13 WARN mapred.JobClient: Use
GenericOptionsParser for parsing the arguments. Applications should
implement Tool for the same.
[exec] 09/10/20 08:45:21 INFO mapred.FileInputFormat: Total input
paths to process : 1
[exec] 09/10/20 08:45:24 INFO mapred.FileInputFormat: Total input
paths to process : 1
[exec] 09/10/20 08:45:25 INFO mapred.MapTask: numReduceTasks: 1
[exec] 09/10/20 08:45:25 INFO mapred.MapTask: io.sort.mb = 100
[exec] 09/10/20 08:45:25 INFO mapred.MapTask: data buffer =
79691776/99614720
[exec] 09/10/20 08:45:25 INFO mapred.MapTask: record buffer =
262144/327680
[exec] 09/10/20 08:45:25 INFO mapred.MapTask: Starting flush of
map output
[exec] 09/10/20 08:45:26 INFO mapred.MapTask: Finished spill 0
[exec] 09/10/20 08:45:26 INFO mapred.TaskRunner:
Task:attempt_local_0001_m_000000_0 is done. And is in the process of
commiting
[exec] 09/10/20 08:45:26 INFO mapred.LocalJobRunner: file:/Users/
xxx/projects/hadoop-jobs/query-hist/data/searches.seq:0+3562
[exec] 09/10/20 08:45:26 INFO mapred.TaskRunner: Task
'attempt_local_0001_m_000000_0' done.
[exec] 09/10/20 08:45:26 INFO mapred.LocalJobRunner:
[exec] 09/10/20 08:45:26 INFO mapred.Merger: Merging 1 sorted
segments
[exec] 09/10/20 08:45:26 INFO mapred.Merger: Down to the last
merge-pass, with 1 segments left of total size: 2641 bytes
[exec] 09/10/20 08:45:26 INFO mapred.LocalJobRunner:
[exec] 09/10/20 08:45:27 INFO mapred.TaskRunner:
Task:attempt_local_0001_r_000000_0 is done. And is in the process of
commiting
[exec] 09/10/20 08:45:27 INFO mapred.LocalJobRunner:
[exec] 09/10/20 08:45:27 INFO mapred.TaskRunner: Task
attempt_local_0001_r_000000_0 is allowed to commit now
[exec] 09/10/20 08:45:28 INFO mapred.FileOutputCommitter: Saved
output of task 'attempt_local_0001_r_000000_0' to
hdfs://master0:54310/tmp/hadoop-xxx/import_42309_9506481
[exec] 09/10/20 08:45:28 INFO mapred.LocalJobRunner: reduce >
reduce
[exec] 09/10/20 08:45:28 INFO mapred.TaskRunner: Task
'attempt_local_0001_r_000000_0' done.
[exec] 09/10/20 08:45:29 INFO flow.FlowStep: [import] starting
step: (2/3) TempHfs["SequenceFile[['search_term', 'term_count']]"]
[import/16662/]
[exec] 09/10/20 08:45:30 WARN mapred.JobClient: Use
GenericOptionsParser for parsing the arguments. Applications should
implement Tool for the same.
[exec] 09/10/20 08:45:38 INFO mapred.FileInputFormat: Total input
paths to process : 1
[exec] 09/10/20 08:45:42 WARN flow.FlowStep: [import] completion
events count: 0
[exec] 09/10/20 08:45:42 WARN flow.FlowStep: [import] abandoning
step: (3/3) Lfs["TextLine[['offset', 'line']->[ALL]]"]["output"]"],
predecessor failed: (2/3) TempHfs["SequenceFile[['search_term',
'term_count']]"][import/16662/]
[exec] 09/10/20 08:45:42 INFO flow.FlowStep: [import] stopping:
(3/3) Lfs["TextLine[['offset', 'line']->[ALL]]"]["output"]"]
[exec] 09/10/20 08:45:42 WARN flow.Flow: stopping jobs
[exec] 09/10/20 08:45:42 INFO flow.FlowStep: [import] stopping:
(3/3) Lfs["TextLine[['offset', 'line']->[ALL]]"]["output"]"]
[exec] 09/10/20 08:45:42 INFO flow.FlowStep: [import] stopping:
(2/3) TempHfs["SequenceFile[['search_term', 'term_count']]"][import/
16662/]
[exec] 09/10/20 08:45:42 INFO flow.FlowStep: [import] stopping:
(1/3) TempHfs["SequenceFile[['timestamp', 'visitor_id', 'search_term',
'request_id', 'ts']]"][import/42309/]
[exec] 09/10/20 08:45:42 WARN flow.Flow: stopped jobs
[exec] 09/10/20 08:45:42 WARN flow.Flow: shutting down job
executor
[exec] 09/10/20 08:45:42 WARN flow.Flow: shutdown complete
[exec] 09/10/20 08:45:43 INFO hadoop.Hadoop18TapUtil: deleting
temp path output/_temporary
[exec] Exception in thread "main" cascading.flow.FlowException:
step failed: (2/3) TempHfs["SequenceFile[['search_term',
'term_count']]"][import/16662/]
[exec] at cascading.flow.FlowStepJob.blockOnJob
(FlowStepJob.java:168)
[exec] at cascading.flow.FlowStepJob.start(FlowStepJob.java:
131)
[exec] at cascading.flow.FlowStepJob.call(FlowStepJob.java:
120)
[exec] at cascading.flow.FlowStepJob.call(FlowStepJob.java:
39)
[exec] at java.util.concurrent.FutureTask$Sync.innerRun
(FutureTask.java:303)
[exec] at java.util.concurrent.FutureTask.run(FutureTask.java:
138)
[exec] at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
[exec] at java.util.concurrent.ThreadPoolExecutor$Worker.run
(ThreadPoolExecutor.java:908)
[exec] at java.lang.Thread.run(Thread.java:637)
[exec] Result: 1

[Nate Murray]

This is terribly embarrassing.

The problem wasn't cascading at all. I ran out of disk space on the
root partition of my Hadoop master.

Sorry for posting here too quickly,

Nate

[Chris K Wensel]

good news (for me). thanks for following up so quickly.

cheers,
chris

--
Chris K Wensel
ch...@concurrentinc.com
http://www.concurrentinc.com