Kryo Serialization breaks with HashJoin, but not CoGroup

[jd]

I'm not sure if this is a bug or user error or what.. but I can get Kryo serialization to break when using the HashJoin, but when using the CoGroup, it works fine.

cascading.CascadingException: unable to load serializer for: java.util.ArrayList from: org.apache.hadoop.io.serializer.SerializationFactory

at cascading.tuple.hadoop.TupleSerialization.getNewSerializer(TupleSerialization.java:420)

at cascading.tuple.hadoop.TupleSerialization$SerializationElementWriter.write(TupleSerialization.java:695)

at cascading.tuple.io.TupleOutputStream.write(TupleOutputStream.java:102)

at cascading.tuple.io.TupleOutputStream.writeTuple(TupleOutputStream.java:64)

at cascading.tuple.collect.SpillableTupleList.writeList(SpillableTupleList.java:347)

at cascading.tuple.collect.SpillableTupleList.doSpill(SpillableTupleList.java:300)

at cascading.tuple.collect.SpillableTupleList.add(SpillableTupleList.java:270)

at cascading.tuple.collect.SpillableTupleList.add(SpillableTupleList.java:54)

at cascading.flow.stream.MemoryHashJoinGate.receive(MemoryHashJoinGate.java:79)

at cascading.flow.stream.MemoryHashJoinGate.receive(MemoryHashJoinGate.java:37)

at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:67)

at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:93)

at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:86)

at cascading.operation.Insert.operate(Insert.java:54)

at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:86)

at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:38)

at cascading.flow.stream.SourceStage.map(SourceStage.java:102)

at cascading.flow.stream.SourceStage.run(SourceStage.java:58)

at cascading.flow.hadoop.FlowMapper.run(FlowMapper.java:112)

at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:358)

at org.apache.hadoop.mapred.MapTask.run(MapTask.java:307)

at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:177)

    @Test

    /**

     * Test the FilteredCountBy aggregator

     */

    public void testSerializeBreaks() throws Exception {

        Pipe inPipe1     = new Pipe("inPipe1");

        Pipe inPipe2     = new Pipe("inPipe2");

        Tap inTap1  = getTextFileTap("/tmp/test/in1", Fields.ALL, "\t", null );

        Tap inTap2  = getTextFileTap("/tmp/test/in2", Fields.ALL, "\t", null );

        Tap outTap  = getTextFileTap("/tmp/test/out", Fields.ALL, "\t", null );

        Properties properties = new Properties();

        properties.setProperty("io.serializations", "cascading.kryo.KryoSerialization");

        AppProps.setApplicationJarClass(properties, AggregateTest.class);

        HadoopFlowConnector myFlowConnector = new HadoopFlowConnector( properties );

        Pipe lhs = new Each(inPipe1,new Insert(new Fields("I","J"),1,3) );

        ArrayList<Object> testList = new ArrayList<Object>();

        testList.add("foo");

        testList.add(1);

        testList.add(1.034);

        Pipe rhs = new Each(inPipe2,new Insert(new Fields("M","N"),1,testList) );

        Pipe joined = new HashJoin(lhs,new Fields("I"),rhs,new Fields("M"),new InnerJoin() );

        HadoopFlow transFlow = (HadoopFlow)myFlowConnector.connect( "test",

                                                  Cascades.tapsMap(

                                                          Pipe.pipes(inPipe1, inPipe2), Tap.taps(inTap1, inTap2)),

                                                  outTap,

                                                  joined );

        transFlow.complete();

        TupleEntryIterator iter =  transFlow.openSink();

        while( iter.hasNext() ) {

            TupleEntry entry = iter.next();

            System.out.println(" " + entry );

        }

    }

    @Test

    /**

     * Test the FilteredCountBy aggregator

     */

    public void testSerializeWorks() throws Exception {

        Pipe inPipe1     = new Pipe("inPipe1");

        Pipe inPipe2     = new Pipe("inPipe2");

        Tap inTap1  = getTextFileTap("/tmp/test/in1", Fields.ALL, "\t", null );

        Tap inTap2  = getTextFileTap("/tmp/test/in2", Fields.ALL, "\t", null );

        Tap outTap  = getTextFileTap("/tmp/test/out", Fields.ALL, "\t", null );

        Properties properties = new Properties();

        properties.setProperty("io.serializations", "cascading.kryo.KryoSerialization");

        AppProps.setApplicationJarClass(properties, AggregateTest.class);

        HadoopFlowConnector myFlowConnector = new HadoopFlowConnector( properties );

        Pipe lhs = new Each(inPipe1,new Insert(new Fields("I","J"),1,3) );

        ArrayList<Object> testList = new ArrayList<Object>();

        testList.add("foo");

        testList.add(1);

        testList.add(1.034);

        Pipe rhs = new Each(inPipe2,new Insert(new Fields("M","N"),1,testList) );

        Pipe joined = new CoGroup(lhs,new Fields("I"),rhs,new Fields("M"),new InnerJoin() );

        HadoopFlow transFlow = (HadoopFlow)myFlowConnector.connect( "test",

                                                  Cascades.tapsMap(

                                                          Pipe.pipes(inPipe1, inPipe2), Tap.taps(inTap1, inTap2)),

                                                  outTap,

                                                  joined );

        transFlow.complete();

        TupleEntryIterator iter =  transFlow.openSink();

        while( iter.hasNext() ) {

            TupleEntry entry = iter.next();

            System.out.println(" " + entry );

        }

    }

    @SuppressWarnings ({"unchecked"})

    public static Tap getTextFileTap ( String fullFilename,

                                       Fields fields,

                                       String delimiter,

                                       String quote )

    {

        // Scheme to be used.

        Scheme sourceScheme =

                new TextDelimited(

                        fields,

                        null, /* Compression */

                        false, /* skip header */

                        false, /* write header */

                        delimiter,

                        true, /* strict - Must have expected number of fields */

                        quote,

                        null, // types

                        true /* safe */

                        );

        // Build the tap.

        return new Hfs( sourceScheme, fullFilename, SinkMode.REPLACE );

    }

[Chris K Wensel]

Make sure you are using the latest wip...

commit 1114078252cc407b0895475e67900bd757f569cb

Author: Chris K Wensel <ch...@wensel.net>

Date:   Mon Apr 30 14:27:23 2012 -0700

    fixed issue where TupleSerialization was not seeing the current properties config in some cases

--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To view this discussion on the web visit https://groups.google.com/d/msg/cascading-user/-/FsY2M7tmmjwJ.
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

[jd]

Thanks.

To post to this group, send email to cascading-user@googlegroups.com.
To unsubscribe from this group, send email to cascading-user+unsubscribe@googlegroups.com.

For more options, visit this group at http://groups.google.com/group/cascading-user?hl=en.