testing word count example in cascading 2.0 (wip 306)

[Ted Naleid]

I've been trying to upgrade to Cascading 2 and have been hitting some issues.  My canary-in-a-coal-mine is the Word Count example in the cascading documentation.

I have a groovy test that worked against the Cascading 1.2.6 build but is failing in 2 because of an error complaining that a JobConf can't be cast to java.util.Properties.  

I think it's because my test is using a `Lfs` source file which I'm guessing doesn't work with the new HadoopFlowConnector; in the conversion I needed to change from a `FlowConnector` (which was ok in 1.2.6) to a `HadoopFlowConnector`.

Are those incompatible in Cascading 2?   Is there any way to use a `Lfs` source with the `HadoopFlowConnector` or do I need to stick it in hadoop and use an `Hfs`?

-Ted

P.S.  This is with hadoop-0.20.2-cdh3u3 using cascading 2.0 wip 306.   Here's the stacktrace:

cascading.flow.planner.PlannerException: could not build flow from assembly: [org.apache.hadoop.mapred.JobConf cannot be cast to java.util.Properties]

at cascading.flow.planner.FlowPlanner.handleExceptionDuringPlanning(FlowPlanner.java:502)

at cascading.flow.hadoop.planner.HadoopPlanner.buildFlow(HadoopPlanner.java:230)

at cascading.flow.FlowConnector.connect(FlowConnector.java:454)

at cascading.flow.FlowConnector.connect(FlowConnector.java:445)

at cascading.flow.FlowConnector.connect(FlowConnector.java:421)

at cascading.flow.FlowConnector.connect(FlowConnector.java:270)

at cascading.flow.FlowConnector.connect(FlowConnector.java:215)

at org.cascading.example.WordCount.createWorkflow(WordCount.java:46)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)

at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:90)

at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite$StaticMetaMethodSiteNoUnwrapNoCoerce.invoke(StaticMetaMethodSite.java:148)

at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite.call(StaticMetaMethodSite.java:88)

at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:42)

at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:108)

at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:120)

at org.cascading.example.WordCountTests.testWordCountTinyDoc(WordCountTests.groovy:26)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)

at org.junit.internal.runners.JUnit38ClassRunner.run(JUnit38ClassRunner.java:79)

at org.junit.runner.JUnitCore.run(JUnitCore.java:137)

at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:63)

Caused by: java.lang.ClassCastException: org.apache.hadoop.mapred.JobConf cannot be cast to java.util.Properties

at cascading.scheme.local.TextLine.sourceConfInit(TextLine.java:59)

at cascading.tap.Tap.sourceConfInit(Tap.java:178)

at cascading.tap.hadoop.Hfs.sourceConfInit(Hfs.java:367)

at cascading.tap.hadoop.Hfs.sourceConfInit(Hfs.java:78)

at cascading.flow.hadoop.HadoopFlowStep.initFromSources(HadoopFlowStep.java:326)

at cascading.flow.hadoop.HadoopFlowStep.getInitializedConfig(HadoopFlowStep.java:96)

at cascading.flow.hadoop.HadoopFlowStep.createFlowStepJob(HadoopFlowStep.java:196)

at cascading.flow.hadoop.HadoopFlowStep.createFlowStepJob(HadoopFlowStep.java:66)

at cascading.flow.planner.BaseFlowStep.getFlowStepJob(BaseFlowStep.java:514)

at cascading.flow.BaseFlow.initializeNewJobsMap(BaseFlow.java:1181)

at cascading.flow.BaseFlow.initialize(BaseFlow.java:184)

at cascading.flow.hadoop.planner.HadoopPlanner.buildFlow(HadoopPlanner.java:224)

... 34 more

Here's the org.cascading.example.WordCount.java file that's pretty much a copy of the example in the documentation:

package org.cascading.example;

import cascading.flow.Flow;

import cascading.flow.FlowConnector;

import cascading.flow.hadoop.HadoopFlowConnector;

import cascading.operation.Aggregator;

import cascading.operation.Debug;

import cascading.operation.Function;

import cascading.operation.aggregator.Count;

import cascading.operation.regex.RegexGenerator;

import cascading.pipe.Each;

import cascading.pipe.Every;

import cascading.pipe.GroupBy;

import cascading.pipe.Pipe;

import cascading.property.AppProps;

import cascading.scheme.Scheme;

import cascading.scheme.hadoop.TextLine;

import cascading.tap.SinkMode;

import cascading.tap.Tap;

import cascading.tap.hadoop.Hfs;

import cascading.tuple.Fields;

import java.util.Properties;

public class WordCount {

    protected static Flow createWorkflow(Tap source, Tap sink) {

        Pipe assembly = new Pipe( "wordcount" );

        String regex = "(?<!\\pL)(?=\\pL)[^ ]*(?<=\\pL)(?!\\pL)";

        Function function = new RegexGenerator( new Fields( "word" ), regex );

        assembly = new Each( assembly, new Fields( "line" ), function );

        assembly = new Each( assembly, new Debug(Debug.Output.STDOUT) );

        assembly = new GroupBy( assembly, new Fields( "word" ) );

        Aggregator count = new Count( new Fields( "count" ) );

        assembly = new Every( assembly, count );

        Properties properties = new Properties();

        AppProps.setApplicationJarClass(properties, WordCount.class);

        FlowConnector flowConnector = new HadoopFlowConnector(properties);

        Flow flow = flowConnector.connect( "word-count", source, sink, assembly );

        

        return flow;

    }

    public static void main(String[] args) {

        String inputPath = args[0];

        String outputPath = args[1];

        

        Scheme sourceScheme = new TextLine( new Fields( "line" ) );

        Tap source = new Hfs( sourceScheme, inputPath );

        Scheme sinkScheme = new TextLine( new Fields( "word", "count" ) );

        Tap sink = new Hfs( sinkScheme, outputPath, SinkMode.REPLACE );

        

        Flow flow = createWorkflow(source, sink);

        flow.complete();

        //flow.writeDOT("build/flow.dot");

    }

}

And this is the org.cascading.example.WordCountTests.groovy test:

package org.cascading.example

import cascading.CascadingTestCase

import cascading.tap.Tap

import cascading.flow.Flow

import cascading.tuple.TupleEntryIterator

import cascading.tap.hadoop.Lfs

import cascading.scheme.local.TextLine

class WordCountTests extends CascadingTestCase {

    String outputPath = "build/test/output/wordcount/";

    public WordCountTests() {

        super( "word count tests" );

    }

    public void testWordCountTinyDoc() throws IOException {

        String contents = """

            The Lazy Dog Jumped

            Over The Lazy Dog

        """

        Tap source = new Lfs( new TextLine(), createTempFile(contents))

        Tap sink = new Lfs( new TextLine(), outputPath + "tinyDoc", true )

        

        Flow flow = WordCount.createWorkflow(source, sink)   // THIS IS THE LINE THATS FAILING IN THE TEST

        flow.complete();

        //flow.writeDOT("build/wordCountFlow.dot")

        validateLength(flow, 5)

        

        Map results = flowResultsMap(flow)

        

        results.The == ["2"]

        results.Lazy == ["2"]

        results.Dog == ["2"]

        results.Jumped == ["1"]

        results.Over == ["1"]

    }

    

    String createTempFile(String contents, String name = "tempfile", String extension = ".txt") {

        File file = File.createTempFile(name, extension)

        file.deleteOnExit()

        file << contents

        return file.absolutePath

    }

    

    Map flowResultsMap(flow) {

        TupleEntryIterator iterator = flow.openSink()

        

        return iterator.inject([:]) { map, result ->

            def fields = result.get(1).split("\t")

            map[fields[0]] = fields[1..-1]

            return map

        }

    }

}

[Chris K Wensel]

This isn't what's compatible..

cascading.scheme.local.TextLine.sourceConfInit(TextLine.java:59)

You need to use the TextLine in the hadoop package. 

if you aren't using local mode, you can leave out the cascading-local jar from your classpath.

ckw

--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To post to this group, send email to cascadi...@googlegroups.com.
To unsubscribe from this group, send email to cascading-use...@googlegroups.com.
For more options, visit this group at http://groups.google.com/group/cascading-user?hl=en.

--

Chris K Wensel

ch...@concurrentinc.com

http://concurrentinc.com

[Ted Naleid]

Thanks for the quick response Chris.  Switching to the TextLine in the hadoop package instead of the local package fixed it and made it work great with Lfs.  The test below passes with that change.

-Ted