MapReduceFlow doesn't Topological Sort as expected because Taps use a NullScheme

[Nate Murray]

Greetings, I think this may be a bug. (branch: wip-1.1)

It appears that MapReduceFlow doesn't topological sort as expected
because Taps use a NullScheme.

For instance, say I have `flow1` which uses "normal" Taps for sources
and sinks. I then want the sink of `flow1` to be the source of my
MapReduceFlow. Currently MapReduceFlow only allows you to pass in file
paths; from which it creates its own source and sink Taps (via
MapReduceFlow#createSources and #createSinks). However, these Taps
have a NullScheme().

CascadeConnector#makeTapGraph adds the Tap's created in the
MapReduceFlow to the graph, but because Tap#equals requires that the
schemes be equal the topological sort sees them as different and
therefore independent. The jobs try to run simultaneously and fail.

It appears to me that we could create a new constructor for
MapReduceFlow that would accept "normal" Taps for source and sink and
then set the paths on the jobConf via tap.getPath() (within
MapReduceFlow).

Do you think this would work? I realize you probably created the
NullScheme for a reason, but I'm not yet clear on why.

Thanks,

Nate

[Chris K Wensel]

Likely a bug. I'll take a look at it as soon as I can. This would
affect 1.0 as well.

ckw

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

[Nate Murray]

Here are more notes after looking into this.

I thought that maybe one way to approach this is to have a TempHFs
between `flow1` and the MR job. The thought was this _could_ work
because TempHfs has an isNull option, which creates a NullScheme.
Since I have a NullScheme in my MR job hopefully the Taps would line
up in when generating the dependency graph.

But not so fast, TempHfs's NullScheme *and* MapReduceFlow's NullScheme
are both inner classes. So again, Tap#equals says "not equal" b/c the
classes really aren't the same. Does it make sense to create a
NullScheme.java and unify the NullScheme in these two places?

Nate

[Chris K Wensel]

re NullScheme. i need to look at the code, but they are not likely
semantically equivalent. but if they are, breaking it out could be
useful.

ckw

[Chris K Wensel]

I can't repro this in 1.0. will try 1.1 later this weekend.

NullScheme in this context subclasses Scheme, which would return
equals() == true in all cases. so I don't think that is the issue.

there could be an issue if one file path is relative and the same file
path is absolute in another job.

ckw

On Oct 23, 2009, at 3:02 PM, Nate Murray wrote:

>

[Nate Murray]

Chris,

Attached is a patch for MapReduceFlow.java that allows a MapReduce
flow to explicitly set Taps. This commit is also at github at:
http://github.com/jashmenn/cascading/commit/90c119261826740a2008c111e12c26526086b281

A couple notes
1) I've kept the interface for the constructor that creates Taps from
the JobConf
2) I've added a convenience constructor for simply passing in a single
source and sink. In my mind, this is a common case.
3) The designated initializer now becomes the more general case of
passing in a Map<String, Tap> of sources, sinks, and traps.

Let me know what you think about this and if you want to make any
changes,

Nate

From 90c119261826740a2008c111e12c26526086b281 Mon Sep 17 00:00:00 2001
From: Nate Murray <na...@natemurray.com>
Date: Fri, 23 Oct 2009 15:54:01 -0700
Subject: [PATCH] updated MapReduceFlow to accept explicit Taps

---
src/core/cascading/flow/MapReduceFlow.java | 54 ++++++++++++++++++++
+++++--
1 files changed, 50 insertions(+), 4 deletions(-)

diff --git a/src/core/cascading/flow/MapReduceFlow.java b/src/core/
cascading/flow/MapReduceFlow.java
index e777140..3597d10 100644
--- a/src/core/cascading/flow/MapReduceFlow.java
+++ b/src/core/cascading/flow/MapReduceFlow.java
@@ -96,7 +96,7 @@ public class MapReduceFlow extends Flow
*/
public MapReduceFlow( String name, JobConf jobConf, boolean
deleteSinkOnInit )
{
- this( name, jobConf, deleteSinkOnInit, true );
+ this( name, jobConf, null, null, null, deleteSinkOnInit, true );
}

/**
@@ -109,13 +109,59 @@ public class MapReduceFlow extends Flow
*/
public MapReduceFlow( String name, JobConf jobConf, boolean
deleteSinkOnInit, boolean stopJobsOnExit )
{
+ this( name, jobConf, null, null, null, deleteSinkOnInit,
stopJobsOnExit );
+ this.setSources( createSources( jobConf ) );
+ this.setSinks( createSinks( jobConf ) );
+ this.setTraps( createTraps( jobConf ) );
+ }
+
+ /**
+ * Constructor MapReduceFlow creates a new MapReduceFlow instance.
+ *
+ * @param name of type String
+ * @param jobConf of type JobConf
+ * @param source of type Tap
+ * @param sink of type Tap
+ * @param deleteSinkOnInit of type boolean
+ * @param stopJobsOnExit of type boolean
+ */
+ public MapReduceFlow( String name, JobConf jobConf, Tap source, Tap
sink, boolean deleteSinkOnInit, boolean stopJobsOnExit )
+ {
+ this( name, jobConf, null, null, null, deleteSinkOnInit,
stopJobsOnExit );
+
+ Map<String, Tap> sources = new HashMap<String, Tap>();
+ Map<String, Tap> sinks = new HashMap<String, Tap>();
+ Map<String, Tap> traps = new HashMap<String, Tap>();
+
+ sources.put( source.getPath().toString(), source );
+ sinks.put( sink.getPath().toString(), sink );
+
+ this.setSources( sources );
+ this.setSinks( sinks );
+ this.setTraps( traps );
+ }
+
+ /**
+ * Constructor MapReduceFlow creates a new MapReduceFlow instance.
+ *
+ * @param name of type String
+ * @param jobConf of type JobConf
+ * @param sources of type Map<String, Tap>
+ * @param sinks of type Map<String, Tap>
+ * @param traps of type Map<String, Tap>
+ * @param deleteSinkOnInit of type boolean
+ * @param stopJobsOnExit of type boolean
+ */
+ public MapReduceFlow( String name, JobConf jobConf, Map<String,
Tap> sources, Map<String, Tap> sinks, Map<String, Tap> traps,
+ boolean deleteSinkOnInit, boolean stopJobsOnExit )
+ {
this.deleteSinkOnInit = deleteSinkOnInit;
this.stopJobsOnExit = stopJobsOnExit;

setName( name );
- setSources( createSources( jobConf ) );
- setSinks( createSinks( jobConf ) );
- setTraps( createTraps( jobConf ) );
+ setSources( sources );
+ setSinks( sinks );
+ setTraps( traps );
setStepGraph( makeStepGraph( jobConf ) );
}

--
1.6.2.4

[Nate Murray]

Actually, scratch that last patch. This one replaces it. Fixes a bug
with the null taps. Github: http://github.com/jashmenn/cascading/commit/f0dd84cd89da70c326e7285034e982c33d2d7388

From f0dd84cd89da70c326e7285034e982c33d2d7388 Mon Sep 17 00:00:00 2001

From: Nate Murray <na...@natemurray.com>
Date: Fri, 23 Oct 2009 15:54:01 -0700

Subject: [PATCH] updated MapReduceFlow to accept explicit Taps (second
try)

---
src/core/cascading/flow/MapReduceFlow.java | 64 ++++++++++++++++++++
++++++--
1 files changed, 60 insertions(+), 4 deletions(-)

diff --git a/src/core/cascading/flow/MapReduceFlow.java b/src/core/
cascading/flow/MapReduceFlow.java

index e777140..9e4bb14 100644
--- a/src/core/cascading/flow/MapReduceFlow.java
+++ b/src/core/cascading/flow/MapReduceFlow.java
@@ -109,14 +109,70 @@ public class MapReduceFlow extends Flow

*/
public MapReduceFlow( String name, JobConf jobConf, boolean
deleteSinkOnInit, boolean stopJobsOnExit )
{
+ this( name, jobConf, null, null, null, deleteSinkOnInit,
stopJobsOnExit );

+ setSources( createSources( jobConf ) );
+ setSinks( createSinks( jobConf ) );
+ setTraps( createTraps( jobConf ) );
+ setStepGraph( makeStepGraph( jobConf ) );

+ setSources( sources );
+ setSinks( sinks );
+ setTraps( traps );

+ setStepGraph( makeStepGraph( jobConf ) );

+ }
+
+ /**
+ * Constructor MapReduceFlow creates a new MapReduceFlow instance.
+ *
+ * @param name of type String
+ * @param jobConf of type JobConf
+ * @param sources of type Map<String, Tap>
+ * @param sinks of type Map<String, Tap>
+ * @param traps of type Map<String, Tap>
+ * @param deleteSinkOnInit of type boolean
+ * @param stopJobsOnExit of type boolean
+ */
+ public MapReduceFlow( String name, JobConf jobConf, Map<String,
Tap> sources, Map<String, Tap> sinks, Map<String, Tap> traps,
+ boolean deleteSinkOnInit, boolean stopJobsOnExit )
+ {
this.deleteSinkOnInit = deleteSinkOnInit;
this.stopJobsOnExit = stopJobsOnExit;

setName( name );
- setSources( createSources( jobConf ) );
- setSinks( createSinks( jobConf ) );
- setTraps( createTraps( jobConf ) );

- setStepGraph( makeStepGraph( jobConf ) );
+
+ if( sources != null )

+ setSources( sources );
+

+ if( sinks != null )

+ setSinks( sinks );
+

+ if( traps != null )
+ setTraps( traps );
+
+ if( (sources != null) && (sinks != null) && (traps != null) )
+ setStepGraph( makeStepGraph( jobConf ) );
}

private StepGraph makeStepGraph( JobConf jobConf )
--
1.6.2.4

On Oct 26, 11:37 am, Nate Murray <nmur...@attinteractive.com> wrote:
> Chris,
>
> Attached is a patch for MapReduceFlow.java that allows a MapReduce

> flow to explicitly set Taps. This commit is also at github at:http://github.com/jashmenn/cascading/commit/90c119261826740a2008c111e...

>
> A couple notes
> 1) I've kept the interface for the constructor that creates Taps from
> the JobConf
> 2) I've added a convenience constructor for simply passing in a single
> source and sink. In my mind, this is a common case.
> 3) The designated initializer now becomes the more general case of
> passing in a Map<String, Tap> of sources, sinks, and traps.
>
> Let me know what you think about this and if you want to make any
> changes,
>
> Nate
>
> From 90c119261826740a2008c111e12c26526086b281 Mon Sep 17 00:00:00 2001

> From: Nate Murray <n...@natemurray.com>

[Chris K Wensel]

Thanks. I'll try and give this a look soon.

cheers,
chris