You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@pig.apache.org by Jonathan Coveney <jc...@gmail.com> on 2011/01/25 19:24:51 UTC
Unexpected data type -1 found in stream.
I've been able to isolate the problem, but have no idea what is causing it.
The input is in this form (this is correct):
{({(a),(b),(c)}),({(a),(b),(c)}),({(a),(b),(c)})}
and the output is in this form:
{(b,c,3),(c,a,3),(b,a,3)}
which is also correct. By placing prints and whatnot, I can see that the
error is coming once I return the second bag.
public DataBag exec(Tuple input) throws IOException {
try {
accumulate(input);
DataBag bag = getValue();
System.out.println(input.get(0).toString());
System.out.println(bag.toString());
return bag;
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (exec)
" + this.getClass().getSimpleName();
throw new ExecException(msg, errCode,
PigException.BUG, e);
}
}
The prints are how I saw that it calculated properly, and I know it's not an
error within exec because it's not throwing an exception. So something weird
is going on afterwards.
It'd be great to understand what is going on, because I think this is what
was plaguing an algebraic version of another script...
Is there something special you have to do if the form of your output is
significantly different from the form of your input? Here is the script that
generates this:
register /path/to/myudf.jar;
A = LOAD 'test.txt' as (a:chararray, b:chararray);
B = GROUP A BY a;
C = FOREACH B GENERATE A.b;
D = GROUP C ALL;
E = FOREACH D GENERATE myudf.fun.udf(C.b);
So It's weird: I'm getting the output I want, it is a DataBag, I output
that, but something is exploding.
Any ideas what it could be? As always, thanks.
Here's from grunt:
java.io.IOException: java.lang.RuntimeException: Unexpected data type -1
found in stream.
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.runPipeline(PigMapReduce.java:438)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.processOnePackageOutput(PigMapReduce.java:401)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:381)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:251)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
at
org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:566)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:408)
at
org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:216)
Caused by: java.lang.RuntimeException: Unexpected data type -1 found in
stream.
at
org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:478)
at
org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
at
org.apache.pig.data.BinInterSedes.writeBag(BinInterSedes.java:522)
at
org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:361)
at
org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
at
org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:357)
at
org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
at
org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:357)
at
org.apache.pig.impl.io.InterRecordWriter.write(InterRecordWriter.java:73)
at org.apache.pig.impl.io.InterStorage.putNext(InterStorage.java:87)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat$PigRecordWriter.write(PigOutputFormat.java:138)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat$PigRecordWriter.write(PigOutputFormat.java:97)
at
org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:508)
at
org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.runPipeline(PigMapReduce.java:436)
Here's from the logfile:Pig Stack Trace
---------------
ERROR 1066: Unable to open iterator for alias E
org.apache.pig.impl.logicalLayer.FrontendException: ERROR 1066: Unable to
open iterator for alias E
at org.apache.pig.PigServer.openIterator(PigServer.java:754)
at
org.apache.pig.tools.grunt.GruntParser.processDump(GruntParser.java:612)
at
org.apache.pig.tools.pigscript.parser.PigScriptParser.parse(PigScriptParser.java:303)
at
org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:165)
at
org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:141)
at org.apache.pig.tools.grunt.Grunt.run(Grunt.java:76)
at org.apache.pig.Main.run(Main.java:465)
at org.apache.pig.Main.main(Main.java:107)
Caused by: java.io.IOException: Job terminated with anomalous status FAILED
at org.apache.pig.PigServer.openIterator(PigServer.java:744)
... 7 more
================================================================================
Re: Unexpected data type -1 found in stream.
Posted by Charles Gonçalves <ch...@gmail.com>.
done!
I attach the code, script and input data!
On Thu, Feb 10, 2011 at 11:39 AM, Jonathan Coveney <jc...@gmail.com>wrote:
> Charles, I did open an issue here:
> https://issues.apache.org/jira/browse/PIG-1826
> <https://issues.apache.org/jira/browse/PIG-1826>It'd be great if you could
> upload your script and UDF! Hopefully having more vectors will help make it
> easier to figure out what is going on.
>
> 2011/2/10 Charles Gonçalves <ch...@gmail.com>
>
> > Jonathan,
> >
> > did you opened the issue?
> > I'm suffering by the same problem ...
> >
> > Any follow up?
> >
> > On Wed, Jan 26, 2011 at 4:17 PM, Jonathan Coveney <jcoveney@gmail.com
> > >wrote:
> >
> > > I have never really had to raise a bug before, what should I do? Open a
> > > ticket, attach the code and the description, and posit that it may be a
> > > serialiazable error?
> > >
> > > It would make me feel so good if there was a bug, though...I've tried
> to
> > > implement some more elegant algebraic versions of things and gotten the
> > > same
> > > error.
> > >
> > > thanks for taking a look at this
> > >
> > > 2011/1/26 Gianmarco <gi...@gmail.com>
> > >
> > > > Thanks Jonathan,
> > > > your code looks fine.
> > > > I think there is a bug in the serialization code for bags.
> > > >
> > > > From a quick look, the size of the bag is a long but it is read as an
> > > > int when comparing bags.
> > > > Not sure this is the culprit though.
> > > >
> > > > I think you should open a Jira for this.
> > > > --
> > > > Gianmarco De Francisci Morales
> > > >
> > > >
> > > >
> > > > On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com>
> > > > wrote:
> > > > > package squeal.fun;
> > > > > import java.util.Iterator;
> > > > > import java.util.List;
> > > > > import java.util.ArrayList;
> > > > > import java.util.Map;
> > > > > import java.util.HashMap;
> > > > > import java.util.Set;
> > > > > import java.util.HashSet;
> > > > > import java.io.IOException;
> > > > > import org.apache.pig.PigException;
> > > > > import org.apache.pig.backend.executionengine.ExecException;
> > > > > import org.apache.pig.EvalFunc;
> > > > > import org.apache.pig.data.Tuple;
> > > > > import org.apache.pig.data.DataBag;
> > > > > import org.apache.pig.data.BagFactory;
> > > > > import org.apache.pig.data.TupleFactory;
> > > > > import org.apache.pig.impl.util.WrappedIOException;
> > > > > import org.apache.pig.impl.logicalLayer.schema.Schema;
> > > > > import org.apache.pig.data.DataType;
> > > > > import squeal.com.MutableInt;
> > > > >
> > > > > public class numgraph extends EvalFunc<DataBag>{
> > > > > TupleFactory mTupleFactory = TupleFactory.getInstance();
> > > > > BagFactory mBagFactory = BagFactory.getInstance();
> > > > >
> > > > > public DataBag exec(Tuple input) throws IOException {
> > > > > try {
> > > > > accumulate(input);
> > > > > DataBag bag = getValue();
> > > > > System.out.println(input.get(0).toString());
> > > > > System.out.println(bag.toString());
> > > > > return bag;
> > > > >
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs (exec) " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > >
> > > > > public void accumulate(Tuple input) throws IOException {
> > > > > try {
> > > > > buildgraph(input);
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs (accumulate) " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > >
> > > > > //public void cleanup() { thegraph.clear(); }
> > > > >
> > > > > public DataBag getValue() throws IOException {
> > > > > try {
> > > > > return thegraph.toBag();
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs (getValue) " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > >
> > > > > Graph thegraph = null;
> > > > > private class Graph {
> > > > > Map<numpair, MutableInt> graph;
> > > > >
> > > > > Graph() { graph = null; }
> > > > > Graph(Map<numpair,MutableInt> gs) { graph = gs; }
> > > > >
> > > > > Map<numpair,MutableInt> getGraph() { return graph; }
> > > > > void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
> > > > > void inc(numpair look) {
> > > > > MutableInt val = graph.get(look);
> > > > > if (val == null) {
> > > > > val = new MutableInt();
> > > > > graph.put(look,val);
> > > > > } else {
> > > > > val.inc();
> > > > > }
> > > > > }
> > > > >
> > > > > void clear() { graph = null; }
> > > > >
> > > > > @Override
> > > > > public String toString() { return graph.toString(); }
> > > > >
> > > > > void addPairsBag(DataBag c2s) throws IOException {
> > > > > try {
> > > > > List<String> c2list = new ArrayList<String>();
> > > > > for (Tuple tup : c2s) {
> > > > > String cur = (String)tup.get(0);
> > > > > for (String ne : c2list)
> > > > > inc(new numpair(ne, cur));
> > > > > c2list.add(cur);
> > > > > }
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs (addPairsBag) " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > >
> > > > > //This creates a databag in the form of (c2, c2, hits)
> > > > > DataBag toBag() throws IOException {
> > > > > try {
> > > > > DataBag outBag = mBagFactory.newDefaultBag();
> > > > > for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
> > > > > List inList = new ArrayList();
> > > > > Iterator<String> sIt = pairs.getKey().getPartsIt();
> > > > > inList.add(sIt.next()); inList.add(sIt.next());
> > > > > inList.add(pairs.getValue());
> > > > > outBag.add(mTupleFactory.newTuple(inList));
> > > > > }
> > > > > return outBag;
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs (toBag) " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > > }
> > > > > private class numpair {
> > > > > Set<String> pair;
> > > > >
> > > > > numpair(String p1, String p2) {
> > > > > pair = new HashSet<String>(2,1);
> > > > > pair.add(p1);
> > > > > pair.add(p2);
> > > > > }
> > > > > Set<String> getPair() { return pair; }
> > > > > Iterator<String> getPartsIt() { return pair.iterator(); }
> > > > > @Override
> > > > > public boolean equals(Object p) {
> > > > > return p instanceof numpair && ((numpair)p).getPair().equals(pair);
> > > > > }
> > > > > @Override
> > > > > public int hashCode() {
> > > > > return pair.hashCode();
> > > > > }
> > > > >
> > > > > public String toString() { return pair.toString(); }
> > > > > }
> > > > >
> > > > > private void buildgraph(Tuple input) throws IOException {
> > > > > if (input == null || input.size() == 0)
> > > > > return;
> > > > > try {
> > > > > if (thegraph == null)
> > > > > thegraph = new Graph(new HashMap<numpair,MutableInt>());
> > > > > if (thegraph.getGraph() == null)
> > > > > thegraph.setGraph(new HashMap<numpair, MutableInt>());
> > > > > DataBag bag = (DataBag)input.get(0);
> > > > > for (Tuple ne : bag)
> > > > > thegraph.addPairsBag((DataBag)ne.get(0));
> > > > > } catch (ExecException ee) {
> > > > > throw ee;
> > > > > } catch (Exception e) {
> > > > > int errCode = 31415;
> > > > > String msg = "Error while accumulating graphs in " +
> > > > > this.getClass().getSimpleName();
> > > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > > }
> > > > > }
> > > > >
> > > > >
> > > > > @Override
> > > > > public Schema outputSchema(Schema input) {
> > > > > try {
> > > > > Schema bagSchema = new Schema();
> > > > > bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
> > > > > bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
> > > > > bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
> > > > > return new Schema(new
> > > > >
> > >
> Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
> > > > > input), bagSchema, DataType.BAG));
> > > > > } catch (Exception e) {
> > > > > return null;
> > > > > }
> > > > > }
> > > > > }
> > > > >
> > > >
> > >
> >
> >
> >
> > --
> > *Charles Ferreira Gonçalves *
> > http://homepages.dcc.ufmg.br/~charles/
> > UFMG - ICEx - Dcc
> > Cel.: 55 31 87741485
> > Tel.: 55 31 34741485
> > Lab.: 55 31 34095840
> >
>
--
*Charles Ferreira Gonçalves *
http://homepages.dcc.ufmg.br/~charles/
UFMG - ICEx - Dcc
Cel.: 55 31 87741485
Tel.: 55 31 34741485
Lab.: 55 31 34095840
Re: Unexpected data type -1 found in stream.
Posted by Jonathan Coveney <jc...@gmail.com>.
Charles, I did open an issue here:
https://issues.apache.org/jira/browse/PIG-1826
<https://issues.apache.org/jira/browse/PIG-1826>It'd be great if you could
upload your script and UDF! Hopefully having more vectors will help make it
easier to figure out what is going on.
2011/2/10 Charles Gonçalves <ch...@gmail.com>
> Jonathan,
>
> did you opened the issue?
> I'm suffering by the same problem ...
>
> Any follow up?
>
> On Wed, Jan 26, 2011 at 4:17 PM, Jonathan Coveney <jcoveney@gmail.com
> >wrote:
>
> > I have never really had to raise a bug before, what should I do? Open a
> > ticket, attach the code and the description, and posit that it may be a
> > serialiazable error?
> >
> > It would make me feel so good if there was a bug, though...I've tried to
> > implement some more elegant algebraic versions of things and gotten the
> > same
> > error.
> >
> > thanks for taking a look at this
> >
> > 2011/1/26 Gianmarco <gi...@gmail.com>
> >
> > > Thanks Jonathan,
> > > your code looks fine.
> > > I think there is a bug in the serialization code for bags.
> > >
> > > From a quick look, the size of the bag is a long but it is read as an
> > > int when comparing bags.
> > > Not sure this is the culprit though.
> > >
> > > I think you should open a Jira for this.
> > > --
> > > Gianmarco De Francisci Morales
> > >
> > >
> > >
> > > On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com>
> > > wrote:
> > > > package squeal.fun;
> > > > import java.util.Iterator;
> > > > import java.util.List;
> > > > import java.util.ArrayList;
> > > > import java.util.Map;
> > > > import java.util.HashMap;
> > > > import java.util.Set;
> > > > import java.util.HashSet;
> > > > import java.io.IOException;
> > > > import org.apache.pig.PigException;
> > > > import org.apache.pig.backend.executionengine.ExecException;
> > > > import org.apache.pig.EvalFunc;
> > > > import org.apache.pig.data.Tuple;
> > > > import org.apache.pig.data.DataBag;
> > > > import org.apache.pig.data.BagFactory;
> > > > import org.apache.pig.data.TupleFactory;
> > > > import org.apache.pig.impl.util.WrappedIOException;
> > > > import org.apache.pig.impl.logicalLayer.schema.Schema;
> > > > import org.apache.pig.data.DataType;
> > > > import squeal.com.MutableInt;
> > > >
> > > > public class numgraph extends EvalFunc<DataBag>{
> > > > TupleFactory mTupleFactory = TupleFactory.getInstance();
> > > > BagFactory mBagFactory = BagFactory.getInstance();
> > > >
> > > > public DataBag exec(Tuple input) throws IOException {
> > > > try {
> > > > accumulate(input);
> > > > DataBag bag = getValue();
> > > > System.out.println(input.get(0).toString());
> > > > System.out.println(bag.toString());
> > > > return bag;
> > > >
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs (exec) " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > >
> > > > public void accumulate(Tuple input) throws IOException {
> > > > try {
> > > > buildgraph(input);
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs (accumulate) " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > >
> > > > //public void cleanup() { thegraph.clear(); }
> > > >
> > > > public DataBag getValue() throws IOException {
> > > > try {
> > > > return thegraph.toBag();
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs (getValue) " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > >
> > > > Graph thegraph = null;
> > > > private class Graph {
> > > > Map<numpair, MutableInt> graph;
> > > >
> > > > Graph() { graph = null; }
> > > > Graph(Map<numpair,MutableInt> gs) { graph = gs; }
> > > >
> > > > Map<numpair,MutableInt> getGraph() { return graph; }
> > > > void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
> > > > void inc(numpair look) {
> > > > MutableInt val = graph.get(look);
> > > > if (val == null) {
> > > > val = new MutableInt();
> > > > graph.put(look,val);
> > > > } else {
> > > > val.inc();
> > > > }
> > > > }
> > > >
> > > > void clear() { graph = null; }
> > > >
> > > > @Override
> > > > public String toString() { return graph.toString(); }
> > > >
> > > > void addPairsBag(DataBag c2s) throws IOException {
> > > > try {
> > > > List<String> c2list = new ArrayList<String>();
> > > > for (Tuple tup : c2s) {
> > > > String cur = (String)tup.get(0);
> > > > for (String ne : c2list)
> > > > inc(new numpair(ne, cur));
> > > > c2list.add(cur);
> > > > }
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs (addPairsBag) " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > >
> > > > //This creates a databag in the form of (c2, c2, hits)
> > > > DataBag toBag() throws IOException {
> > > > try {
> > > > DataBag outBag = mBagFactory.newDefaultBag();
> > > > for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
> > > > List inList = new ArrayList();
> > > > Iterator<String> sIt = pairs.getKey().getPartsIt();
> > > > inList.add(sIt.next()); inList.add(sIt.next());
> > > > inList.add(pairs.getValue());
> > > > outBag.add(mTupleFactory.newTuple(inList));
> > > > }
> > > > return outBag;
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs (toBag) " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > > }
> > > > private class numpair {
> > > > Set<String> pair;
> > > >
> > > > numpair(String p1, String p2) {
> > > > pair = new HashSet<String>(2,1);
> > > > pair.add(p1);
> > > > pair.add(p2);
> > > > }
> > > > Set<String> getPair() { return pair; }
> > > > Iterator<String> getPartsIt() { return pair.iterator(); }
> > > > @Override
> > > > public boolean equals(Object p) {
> > > > return p instanceof numpair && ((numpair)p).getPair().equals(pair);
> > > > }
> > > > @Override
> > > > public int hashCode() {
> > > > return pair.hashCode();
> > > > }
> > > >
> > > > public String toString() { return pair.toString(); }
> > > > }
> > > >
> > > > private void buildgraph(Tuple input) throws IOException {
> > > > if (input == null || input.size() == 0)
> > > > return;
> > > > try {
> > > > if (thegraph == null)
> > > > thegraph = new Graph(new HashMap<numpair,MutableInt>());
> > > > if (thegraph.getGraph() == null)
> > > > thegraph.setGraph(new HashMap<numpair, MutableInt>());
> > > > DataBag bag = (DataBag)input.get(0);
> > > > for (Tuple ne : bag)
> > > > thegraph.addPairsBag((DataBag)ne.get(0));
> > > > } catch (ExecException ee) {
> > > > throw ee;
> > > > } catch (Exception e) {
> > > > int errCode = 31415;
> > > > String msg = "Error while accumulating graphs in " +
> > > > this.getClass().getSimpleName();
> > > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > > }
> > > > }
> > > >
> > > >
> > > > @Override
> > > > public Schema outputSchema(Schema input) {
> > > > try {
> > > > Schema bagSchema = new Schema();
> > > > bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
> > > > bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
> > > > bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
> > > > return new Schema(new
> > > >
> > Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
> > > > input), bagSchema, DataType.BAG));
> > > > } catch (Exception e) {
> > > > return null;
> > > > }
> > > > }
> > > > }
> > > >
> > >
> >
>
>
>
> --
> *Charles Ferreira Gonçalves *
> http://homepages.dcc.ufmg.br/~charles/
> UFMG - ICEx - Dcc
> Cel.: 55 31 87741485
> Tel.: 55 31 34741485
> Lab.: 55 31 34095840
>
Re: Unexpected data type -1 found in stream.
Posted by Charles Gonçalves <ch...@gmail.com>.
Jonathan,
did you opened the issue?
I'm suffering by the same problem ...
Any follow up?
On Wed, Jan 26, 2011 at 4:17 PM, Jonathan Coveney <jc...@gmail.com>wrote:
> I have never really had to raise a bug before, what should I do? Open a
> ticket, attach the code and the description, and posit that it may be a
> serialiazable error?
>
> It would make me feel so good if there was a bug, though...I've tried to
> implement some more elegant algebraic versions of things and gotten the
> same
> error.
>
> thanks for taking a look at this
>
> 2011/1/26 Gianmarco <gi...@gmail.com>
>
> > Thanks Jonathan,
> > your code looks fine.
> > I think there is a bug in the serialization code for bags.
> >
> > From a quick look, the size of the bag is a long but it is read as an
> > int when comparing bags.
> > Not sure this is the culprit though.
> >
> > I think you should open a Jira for this.
> > --
> > Gianmarco De Francisci Morales
> >
> >
> >
> > On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com>
> > wrote:
> > > package squeal.fun;
> > > import java.util.Iterator;
> > > import java.util.List;
> > > import java.util.ArrayList;
> > > import java.util.Map;
> > > import java.util.HashMap;
> > > import java.util.Set;
> > > import java.util.HashSet;
> > > import java.io.IOException;
> > > import org.apache.pig.PigException;
> > > import org.apache.pig.backend.executionengine.ExecException;
> > > import org.apache.pig.EvalFunc;
> > > import org.apache.pig.data.Tuple;
> > > import org.apache.pig.data.DataBag;
> > > import org.apache.pig.data.BagFactory;
> > > import org.apache.pig.data.TupleFactory;
> > > import org.apache.pig.impl.util.WrappedIOException;
> > > import org.apache.pig.impl.logicalLayer.schema.Schema;
> > > import org.apache.pig.data.DataType;
> > > import squeal.com.MutableInt;
> > >
> > > public class numgraph extends EvalFunc<DataBag>{
> > > TupleFactory mTupleFactory = TupleFactory.getInstance();
> > > BagFactory mBagFactory = BagFactory.getInstance();
> > >
> > > public DataBag exec(Tuple input) throws IOException {
> > > try {
> > > accumulate(input);
> > > DataBag bag = getValue();
> > > System.out.println(input.get(0).toString());
> > > System.out.println(bag.toString());
> > > return bag;
> > >
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs (exec) " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > >
> > > public void accumulate(Tuple input) throws IOException {
> > > try {
> > > buildgraph(input);
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs (accumulate) " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > >
> > > //public void cleanup() { thegraph.clear(); }
> > >
> > > public DataBag getValue() throws IOException {
> > > try {
> > > return thegraph.toBag();
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs (getValue) " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > >
> > > Graph thegraph = null;
> > > private class Graph {
> > > Map<numpair, MutableInt> graph;
> > >
> > > Graph() { graph = null; }
> > > Graph(Map<numpair,MutableInt> gs) { graph = gs; }
> > >
> > > Map<numpair,MutableInt> getGraph() { return graph; }
> > > void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
> > > void inc(numpair look) {
> > > MutableInt val = graph.get(look);
> > > if (val == null) {
> > > val = new MutableInt();
> > > graph.put(look,val);
> > > } else {
> > > val.inc();
> > > }
> > > }
> > >
> > > void clear() { graph = null; }
> > >
> > > @Override
> > > public String toString() { return graph.toString(); }
> > >
> > > void addPairsBag(DataBag c2s) throws IOException {
> > > try {
> > > List<String> c2list = new ArrayList<String>();
> > > for (Tuple tup : c2s) {
> > > String cur = (String)tup.get(0);
> > > for (String ne : c2list)
> > > inc(new numpair(ne, cur));
> > > c2list.add(cur);
> > > }
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs (addPairsBag) " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > >
> > > //This creates a databag in the form of (c2, c2, hits)
> > > DataBag toBag() throws IOException {
> > > try {
> > > DataBag outBag = mBagFactory.newDefaultBag();
> > > for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
> > > List inList = new ArrayList();
> > > Iterator<String> sIt = pairs.getKey().getPartsIt();
> > > inList.add(sIt.next()); inList.add(sIt.next());
> > > inList.add(pairs.getValue());
> > > outBag.add(mTupleFactory.newTuple(inList));
> > > }
> > > return outBag;
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs (toBag) " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > > }
> > > private class numpair {
> > > Set<String> pair;
> > >
> > > numpair(String p1, String p2) {
> > > pair = new HashSet<String>(2,1);
> > > pair.add(p1);
> > > pair.add(p2);
> > > }
> > > Set<String> getPair() { return pair; }
> > > Iterator<String> getPartsIt() { return pair.iterator(); }
> > > @Override
> > > public boolean equals(Object p) {
> > > return p instanceof numpair && ((numpair)p).getPair().equals(pair);
> > > }
> > > @Override
> > > public int hashCode() {
> > > return pair.hashCode();
> > > }
> > >
> > > public String toString() { return pair.toString(); }
> > > }
> > >
> > > private void buildgraph(Tuple input) throws IOException {
> > > if (input == null || input.size() == 0)
> > > return;
> > > try {
> > > if (thegraph == null)
> > > thegraph = new Graph(new HashMap<numpair,MutableInt>());
> > > if (thegraph.getGraph() == null)
> > > thegraph.setGraph(new HashMap<numpair, MutableInt>());
> > > DataBag bag = (DataBag)input.get(0);
> > > for (Tuple ne : bag)
> > > thegraph.addPairsBag((DataBag)ne.get(0));
> > > } catch (ExecException ee) {
> > > throw ee;
> > > } catch (Exception e) {
> > > int errCode = 31415;
> > > String msg = "Error while accumulating graphs in " +
> > > this.getClass().getSimpleName();
> > > throw new ExecException(msg, errCode, PigException.BUG, e);
> > > }
> > > }
> > >
> > >
> > > @Override
> > > public Schema outputSchema(Schema input) {
> > > try {
> > > Schema bagSchema = new Schema();
> > > bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
> > > bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
> > > bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
> > > return new Schema(new
> > >
> Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
> > > input), bagSchema, DataType.BAG));
> > > } catch (Exception e) {
> > > return null;
> > > }
> > > }
> > > }
> > >
> >
>
--
*Charles Ferreira Gonçalves *
http://homepages.dcc.ufmg.br/~charles/
UFMG - ICEx - Dcc
Cel.: 55 31 87741485
Tel.: 55 31 34741485
Lab.: 55 31 34095840
Re: Unexpected data type -1 found in stream.
Posted by Alan Gates <ga...@yahoo-inc.com>.
On Jan 26, 2011, at 10:17 AM, Jonathan Coveney wrote:
> I have never really had to raise a bug before, what should I do?
> Open a
> ticket, attach the code and the description, and posit that it may
> be a
> serialiazable error?
Yes.
>
> It would make me feel so good if there was a bug, though...I've
> tried to
> implement some more elegant algebraic versions of things and gotten
> the same
> error.
Alan.
>
> thanks for taking a look at this
>
> 2011/1/26 Gianmarco <gi...@gmail.com>
>
>> Thanks Jonathan,
>> your code looks fine.
>> I think there is a bug in the serialization code for bags.
>>
>> From a quick look, the size of the bag is a long but it is read as an
>> int when comparing bags.
>> Not sure this is the culprit though.
>>
>> I think you should open a Jira for this.
>> --
>> Gianmarco De Francisci Morales
>>
>>
>>
>> On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com>
>> wrote:
>>> package squeal.fun;
>>> import java.util.Iterator;
>>> import java.util.List;
>>> import java.util.ArrayList;
>>> import java.util.Map;
>>> import java.util.HashMap;
>>> import java.util.Set;
>>> import java.util.HashSet;
>>> import java.io.IOException;
>>> import org.apache.pig.PigException;
>>> import org.apache.pig.backend.executionengine.ExecException;
>>> import org.apache.pig.EvalFunc;
>>> import org.apache.pig.data.Tuple;
>>> import org.apache.pig.data.DataBag;
>>> import org.apache.pig.data.BagFactory;
>>> import org.apache.pig.data.TupleFactory;
>>> import org.apache.pig.impl.util.WrappedIOException;
>>> import org.apache.pig.impl.logicalLayer.schema.Schema;
>>> import org.apache.pig.data.DataType;
>>> import squeal.com.MutableInt;
>>>
>>> public class numgraph extends EvalFunc<DataBag>{
>>> TupleFactory mTupleFactory = TupleFactory.getInstance();
>>> BagFactory mBagFactory = BagFactory.getInstance();
>>>
>>> public DataBag exec(Tuple input) throws IOException {
>>> try {
>>> accumulate(input);
>>> DataBag bag = getValue();
>>> System.out.println(input.get(0).toString());
>>> System.out.println(bag.toString());
>>> return bag;
>>>
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs (exec) " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>>
>>> public void accumulate(Tuple input) throws IOException {
>>> try {
>>> buildgraph(input);
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs (accumulate) " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>>
>>> //public void cleanup() { thegraph.clear(); }
>>>
>>> public DataBag getValue() throws IOException {
>>> try {
>>> return thegraph.toBag();
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs (getValue) " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>>
>>> Graph thegraph = null;
>>> private class Graph {
>>> Map<numpair, MutableInt> graph;
>>>
>>> Graph() { graph = null; }
>>> Graph(Map<numpair,MutableInt> gs) { graph = gs; }
>>>
>>> Map<numpair,MutableInt> getGraph() { return graph; }
>>> void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
>>> void inc(numpair look) {
>>> MutableInt val = graph.get(look);
>>> if (val == null) {
>>> val = new MutableInt();
>>> graph.put(look,val);
>>> } else {
>>> val.inc();
>>> }
>>> }
>>>
>>> void clear() { graph = null; }
>>>
>>> @Override
>>> public String toString() { return graph.toString(); }
>>>
>>> void addPairsBag(DataBag c2s) throws IOException {
>>> try {
>>> List<String> c2list = new ArrayList<String>();
>>> for (Tuple tup : c2s) {
>>> String cur = (String)tup.get(0);
>>> for (String ne : c2list)
>>> inc(new numpair(ne, cur));
>>> c2list.add(cur);
>>> }
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs (addPairsBag) " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>>
>>> //This creates a databag in the form of (c2, c2, hits)
>>> DataBag toBag() throws IOException {
>>> try {
>>> DataBag outBag = mBagFactory.newDefaultBag();
>>> for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
>>> List inList = new ArrayList();
>>> Iterator<String> sIt = pairs.getKey().getPartsIt();
>>> inList.add(sIt.next()); inList.add(sIt.next());
>>> inList.add(pairs.getValue());
>>> outBag.add(mTupleFactory.newTuple(inList));
>>> }
>>> return outBag;
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs (toBag) " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>> }
>>> private class numpair {
>>> Set<String> pair;
>>>
>>> numpair(String p1, String p2) {
>>> pair = new HashSet<String>(2,1);
>>> pair.add(p1);
>>> pair.add(p2);
>>> }
>>> Set<String> getPair() { return pair; }
>>> Iterator<String> getPartsIt() { return pair.iterator(); }
>>> @Override
>>> public boolean equals(Object p) {
>>> return p instanceof numpair && ((numpair)p).getPair().equals(pair);
>>> }
>>> @Override
>>> public int hashCode() {
>>> return pair.hashCode();
>>> }
>>>
>>> public String toString() { return pair.toString(); }
>>> }
>>>
>>> private void buildgraph(Tuple input) throws IOException {
>>> if (input == null || input.size() == 0)
>>> return;
>>> try {
>>> if (thegraph == null)
>>> thegraph = new Graph(new HashMap<numpair,MutableInt>());
>>> if (thegraph.getGraph() == null)
>>> thegraph.setGraph(new HashMap<numpair, MutableInt>());
>>> DataBag bag = (DataBag)input.get(0);
>>> for (Tuple ne : bag)
>>> thegraph.addPairsBag((DataBag)ne.get(0));
>>> } catch (ExecException ee) {
>>> throw ee;
>>> } catch (Exception e) {
>>> int errCode = 31415;
>>> String msg = "Error while accumulating graphs in " +
>>> this.getClass().getSimpleName();
>>> throw new ExecException(msg, errCode, PigException.BUG, e);
>>> }
>>> }
>>>
>>>
>>> @Override
>>> public Schema outputSchema(Schema input) {
>>> try {
>>> Schema bagSchema = new Schema();
>>> bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
>>> bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
>>> bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
>>> return new Schema(new
>>> Schema
>>> .FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
>>> input), bagSchema, DataType.BAG));
>>> } catch (Exception e) {
>>> return null;
>>> }
>>> }
>>> }
>>>
>>
Re: Unexpected data type -1 found in stream.
Posted by Jonathan Coveney <jc...@gmail.com>.
I have never really had to raise a bug before, what should I do? Open a
ticket, attach the code and the description, and posit that it may be a
serialiazable error?
It would make me feel so good if there was a bug, though...I've tried to
implement some more elegant algebraic versions of things and gotten the same
error.
thanks for taking a look at this
2011/1/26 Gianmarco <gi...@gmail.com>
> Thanks Jonathan,
> your code looks fine.
> I think there is a bug in the serialization code for bags.
>
> From a quick look, the size of the bag is a long but it is read as an
> int when comparing bags.
> Not sure this is the culprit though.
>
> I think you should open a Jira for this.
> --
> Gianmarco De Francisci Morales
>
>
>
> On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com>
> wrote:
> > package squeal.fun;
> > import java.util.Iterator;
> > import java.util.List;
> > import java.util.ArrayList;
> > import java.util.Map;
> > import java.util.HashMap;
> > import java.util.Set;
> > import java.util.HashSet;
> > import java.io.IOException;
> > import org.apache.pig.PigException;
> > import org.apache.pig.backend.executionengine.ExecException;
> > import org.apache.pig.EvalFunc;
> > import org.apache.pig.data.Tuple;
> > import org.apache.pig.data.DataBag;
> > import org.apache.pig.data.BagFactory;
> > import org.apache.pig.data.TupleFactory;
> > import org.apache.pig.impl.util.WrappedIOException;
> > import org.apache.pig.impl.logicalLayer.schema.Schema;
> > import org.apache.pig.data.DataType;
> > import squeal.com.MutableInt;
> >
> > public class numgraph extends EvalFunc<DataBag>{
> > TupleFactory mTupleFactory = TupleFactory.getInstance();
> > BagFactory mBagFactory = BagFactory.getInstance();
> >
> > public DataBag exec(Tuple input) throws IOException {
> > try {
> > accumulate(input);
> > DataBag bag = getValue();
> > System.out.println(input.get(0).toString());
> > System.out.println(bag.toString());
> > return bag;
> >
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs (exec) " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> >
> > public void accumulate(Tuple input) throws IOException {
> > try {
> > buildgraph(input);
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs (accumulate) " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> >
> > //public void cleanup() { thegraph.clear(); }
> >
> > public DataBag getValue() throws IOException {
> > try {
> > return thegraph.toBag();
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs (getValue) " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> >
> > Graph thegraph = null;
> > private class Graph {
> > Map<numpair, MutableInt> graph;
> >
> > Graph() { graph = null; }
> > Graph(Map<numpair,MutableInt> gs) { graph = gs; }
> >
> > Map<numpair,MutableInt> getGraph() { return graph; }
> > void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
> > void inc(numpair look) {
> > MutableInt val = graph.get(look);
> > if (val == null) {
> > val = new MutableInt();
> > graph.put(look,val);
> > } else {
> > val.inc();
> > }
> > }
> >
> > void clear() { graph = null; }
> >
> > @Override
> > public String toString() { return graph.toString(); }
> >
> > void addPairsBag(DataBag c2s) throws IOException {
> > try {
> > List<String> c2list = new ArrayList<String>();
> > for (Tuple tup : c2s) {
> > String cur = (String)tup.get(0);
> > for (String ne : c2list)
> > inc(new numpair(ne, cur));
> > c2list.add(cur);
> > }
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs (addPairsBag) " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> >
> > //This creates a databag in the form of (c2, c2, hits)
> > DataBag toBag() throws IOException {
> > try {
> > DataBag outBag = mBagFactory.newDefaultBag();
> > for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
> > List inList = new ArrayList();
> > Iterator<String> sIt = pairs.getKey().getPartsIt();
> > inList.add(sIt.next()); inList.add(sIt.next());
> > inList.add(pairs.getValue());
> > outBag.add(mTupleFactory.newTuple(inList));
> > }
> > return outBag;
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs (toBag) " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> > }
> > private class numpair {
> > Set<String> pair;
> >
> > numpair(String p1, String p2) {
> > pair = new HashSet<String>(2,1);
> > pair.add(p1);
> > pair.add(p2);
> > }
> > Set<String> getPair() { return pair; }
> > Iterator<String> getPartsIt() { return pair.iterator(); }
> > @Override
> > public boolean equals(Object p) {
> > return p instanceof numpair && ((numpair)p).getPair().equals(pair);
> > }
> > @Override
> > public int hashCode() {
> > return pair.hashCode();
> > }
> >
> > public String toString() { return pair.toString(); }
> > }
> >
> > private void buildgraph(Tuple input) throws IOException {
> > if (input == null || input.size() == 0)
> > return;
> > try {
> > if (thegraph == null)
> > thegraph = new Graph(new HashMap<numpair,MutableInt>());
> > if (thegraph.getGraph() == null)
> > thegraph.setGraph(new HashMap<numpair, MutableInt>());
> > DataBag bag = (DataBag)input.get(0);
> > for (Tuple ne : bag)
> > thegraph.addPairsBag((DataBag)ne.get(0));
> > } catch (ExecException ee) {
> > throw ee;
> > } catch (Exception e) {
> > int errCode = 31415;
> > String msg = "Error while accumulating graphs in " +
> > this.getClass().getSimpleName();
> > throw new ExecException(msg, errCode, PigException.BUG, e);
> > }
> > }
> >
> >
> > @Override
> > public Schema outputSchema(Schema input) {
> > try {
> > Schema bagSchema = new Schema();
> > bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
> > bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
> > bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
> > return new Schema(new
> > Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
> > input), bagSchema, DataType.BAG));
> > } catch (Exception e) {
> > return null;
> > }
> > }
> > }
> >
>
Re: Unexpected data type -1 found in stream.
Posted by Gianmarco <gi...@gmail.com>.
Thanks Jonathan,
your code looks fine.
I think there is a bug in the serialization code for bags.
>From a quick look, the size of the bag is a long but it is read as an
int when comparing bags.
Not sure this is the culprit though.
I think you should open a Jira for this.
--
Gianmarco De Francisci Morales
On Tue, Jan 25, 2011 at 21:47, Jonathan Coveney <jc...@gmail.com> wrote:
> package squeal.fun;
> import java.util.Iterator;
> import java.util.List;
> import java.util.ArrayList;
> import java.util.Map;
> import java.util.HashMap;
> import java.util.Set;
> import java.util.HashSet;
> import java.io.IOException;
> import org.apache.pig.PigException;
> import org.apache.pig.backend.executionengine.ExecException;
> import org.apache.pig.EvalFunc;
> import org.apache.pig.data.Tuple;
> import org.apache.pig.data.DataBag;
> import org.apache.pig.data.BagFactory;
> import org.apache.pig.data.TupleFactory;
> import org.apache.pig.impl.util.WrappedIOException;
> import org.apache.pig.impl.logicalLayer.schema.Schema;
> import org.apache.pig.data.DataType;
> import squeal.com.MutableInt;
>
> public class numgraph extends EvalFunc<DataBag>{
> TupleFactory mTupleFactory = TupleFactory.getInstance();
> BagFactory mBagFactory = BagFactory.getInstance();
>
> public DataBag exec(Tuple input) throws IOException {
> try {
> accumulate(input);
> DataBag bag = getValue();
> System.out.println(input.get(0).toString());
> System.out.println(bag.toString());
> return bag;
>
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (exec) " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
>
> public void accumulate(Tuple input) throws IOException {
> try {
> buildgraph(input);
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (accumulate) " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
>
> //public void cleanup() { thegraph.clear(); }
>
> public DataBag getValue() throws IOException {
> try {
> return thegraph.toBag();
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (getValue) " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
>
> Graph thegraph = null;
> private class Graph {
> Map<numpair, MutableInt> graph;
>
> Graph() { graph = null; }
> Graph(Map<numpair,MutableInt> gs) { graph = gs; }
>
> Map<numpair,MutableInt> getGraph() { return graph; }
> void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
> void inc(numpair look) {
> MutableInt val = graph.get(look);
> if (val == null) {
> val = new MutableInt();
> graph.put(look,val);
> } else {
> val.inc();
> }
> }
>
> void clear() { graph = null; }
>
> @Override
> public String toString() { return graph.toString(); }
>
> void addPairsBag(DataBag c2s) throws IOException {
> try {
> List<String> c2list = new ArrayList<String>();
> for (Tuple tup : c2s) {
> String cur = (String)tup.get(0);
> for (String ne : c2list)
> inc(new numpair(ne, cur));
> c2list.add(cur);
> }
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (addPairsBag) " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
>
> //This creates a databag in the form of (c2, c2, hits)
> DataBag toBag() throws IOException {
> try {
> DataBag outBag = mBagFactory.newDefaultBag();
> for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
> List inList = new ArrayList();
> Iterator<String> sIt = pairs.getKey().getPartsIt();
> inList.add(sIt.next()); inList.add(sIt.next());
> inList.add(pairs.getValue());
> outBag.add(mTupleFactory.newTuple(inList));
> }
> return outBag;
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (toBag) " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
> }
> private class numpair {
> Set<String> pair;
>
> numpair(String p1, String p2) {
> pair = new HashSet<String>(2,1);
> pair.add(p1);
> pair.add(p2);
> }
> Set<String> getPair() { return pair; }
> Iterator<String> getPartsIt() { return pair.iterator(); }
> @Override
> public boolean equals(Object p) {
> return p instanceof numpair && ((numpair)p).getPair().equals(pair);
> }
> @Override
> public int hashCode() {
> return pair.hashCode();
> }
>
> public String toString() { return pair.toString(); }
> }
>
> private void buildgraph(Tuple input) throws IOException {
> if (input == null || input.size() == 0)
> return;
> try {
> if (thegraph == null)
> thegraph = new Graph(new HashMap<numpair,MutableInt>());
> if (thegraph.getGraph() == null)
> thegraph.setGraph(new HashMap<numpair, MutableInt>());
> DataBag bag = (DataBag)input.get(0);
> for (Tuple ne : bag)
> thegraph.addPairsBag((DataBag)ne.get(0));
> } catch (ExecException ee) {
> throw ee;
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs in " +
> this.getClass().getSimpleName();
> throw new ExecException(msg, errCode, PigException.BUG, e);
> }
> }
>
>
> @Override
> public Schema outputSchema(Schema input) {
> try {
> Schema bagSchema = new Schema();
> bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
> bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
> bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
> return new Schema(new
> Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
> input), bagSchema, DataType.BAG));
> } catch (Exception e) {
> return null;
> }
> }
> }
>
Re: Unexpected data type -1 found in stream.
Posted by Jonathan Coveney <jc...@gmail.com>.
package squeal.fun;
import java.util.Iterator;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.Set;
import java.util.HashSet;
import java.io.IOException;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.util.WrappedIOException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.data.DataType;
import squeal.com.MutableInt;
public class numgraph extends EvalFunc<DataBag>{
TupleFactory mTupleFactory = TupleFactory.getInstance();
BagFactory mBagFactory = BagFactory.getInstance();
public DataBag exec(Tuple input) throws IOException {
try {
accumulate(input);
DataBag bag = getValue();
System.out.println(input.get(0).toString());
System.out.println(bag.toString());
return bag;
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (exec) " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
public void accumulate(Tuple input) throws IOException {
try {
buildgraph(input);
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (accumulate) " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
//public void cleanup() { thegraph.clear(); }
public DataBag getValue() throws IOException {
try {
return thegraph.toBag();
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (getValue) " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
Graph thegraph = null;
private class Graph {
Map<numpair, MutableInt> graph;
Graph() { graph = null; }
Graph(Map<numpair,MutableInt> gs) { graph = gs; }
Map<numpair,MutableInt> getGraph() { return graph; }
void setGraph(Map<numpair,MutableInt> gs) { graph = gs; }
void inc(numpair look) {
MutableInt val = graph.get(look);
if (val == null) {
val = new MutableInt();
graph.put(look,val);
} else {
val.inc();
}
}
void clear() { graph = null; }
@Override
public String toString() { return graph.toString(); }
void addPairsBag(DataBag c2s) throws IOException {
try {
List<String> c2list = new ArrayList<String>();
for (Tuple tup : c2s) {
String cur = (String)tup.get(0);
for (String ne : c2list)
inc(new numpair(ne, cur));
c2list.add(cur);
}
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (addPairsBag) " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
//This creates a databag in the form of (c2, c2, hits)
DataBag toBag() throws IOException {
try {
DataBag outBag = mBagFactory.newDefaultBag();
for (Map.Entry<numpair,MutableInt> pairs : graph.entrySet()) {
List inList = new ArrayList();
Iterator<String> sIt = pairs.getKey().getPartsIt();
inList.add(sIt.next()); inList.add(sIt.next());
inList.add(pairs.getValue());
outBag.add(mTupleFactory.newTuple(inList));
}
return outBag;
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs (toBag) " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
}
private class numpair {
Set<String> pair;
numpair(String p1, String p2) {
pair = new HashSet<String>(2,1);
pair.add(p1);
pair.add(p2);
}
Set<String> getPair() { return pair; }
Iterator<String> getPartsIt() { return pair.iterator(); }
@Override
public boolean equals(Object p) {
return p instanceof numpair && ((numpair)p).getPair().equals(pair);
}
@Override
public int hashCode() {
return pair.hashCode();
}
public String toString() { return pair.toString(); }
}
private void buildgraph(Tuple input) throws IOException {
if (input == null || input.size() == 0)
return;
try {
if (thegraph == null)
thegraph = new Graph(new HashMap<numpair,MutableInt>());
if (thegraph.getGraph() == null)
thegraph.setGraph(new HashMap<numpair, MutableInt>());
DataBag bag = (DataBag)input.get(0);
for (Tuple ne : bag)
thegraph.addPairsBag((DataBag)ne.get(0));
} catch (ExecException ee) {
throw ee;
} catch (Exception e) {
int errCode = 31415;
String msg = "Error while accumulating graphs in " +
this.getClass().getSimpleName();
throw new ExecException(msg, errCode, PigException.BUG, e);
}
}
@Override
public Schema outputSchema(Schema input) {
try {
Schema bagSchema = new Schema();
bagSchema.add(new Schema.FieldSchema("c2_1",DataType.CHARARRAY));
bagSchema.add(new Schema.FieldSchema("c2_2",DataType.CHARARRAY));
bagSchema.add(new Schema.FieldSchema("hits",DataType.INTEGER));
return new Schema(new
Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(),
input), bagSchema, DataType.BAG));
} catch (Exception e) {
return null;
}
}
}
Re: Unexpected data type -1 found in stream.
Posted by Gianmarco <gi...@gmail.com>.
>From what I see the data type of the DataBag is not correcly recognized.
I guess that that -1 comes from DataType.findType(), that is returning ERROR.
I also assume (but I am not sure) that the concrete type of getValue()
should be AccumulativeBag,
but for some reason it is something different. Maybe the BagFactory is
misbehaving.
Can you post the complete code for your UDF?
Cheers,
--
Gianmarco De Francisci Morales
On Tue, Jan 25, 2011 at 19:24, Jonathan Coveney <jc...@gmail.com> wrote:
> I've been able to isolate the problem, but have no idea what is causing it.
>
> The input is in this form (this is correct):
>
> {({(a),(b),(c)}),({(a),(b),(c)}),({(a),(b),(c)})}
>
> and the output is in this form:
>
> {(b,c,3),(c,a,3),(b,a,3)}
>
> which is also correct. By placing prints and whatnot, I can see that the
> error is coming once I return the second bag.
>
> public DataBag exec(Tuple input) throws IOException {
> try {
> accumulate(input);
> DataBag bag = getValue();
> System.out.println(input.get(0).toString());
> System.out.println(bag.toString());
> return bag;
>
> } catch (Exception e) {
> int errCode = 31415;
> String msg = "Error while accumulating graphs (exec)
> " + this.getClass().getSimpleName();
> throw new ExecException(msg, errCode,
> PigException.BUG, e);
> }
> }
>
> The prints are how I saw that it calculated properly, and I know it's not an
> error within exec because it's not throwing an exception. So something weird
> is going on afterwards.
>
> It'd be great to understand what is going on, because I think this is what
> was plaguing an algebraic version of another script...
>
> Is there something special you have to do if the form of your output is
> significantly different from the form of your input? Here is the script that
> generates this:
>
> register /path/to/myudf.jar;
> A = LOAD 'test.txt' as (a:chararray, b:chararray);
> B = GROUP A BY a;
> C = FOREACH B GENERATE A.b;
> D = GROUP C ALL;
> E = FOREACH D GENERATE myudf.fun.udf(C.b);
>
> So It's weird: I'm getting the output I want, it is a DataBag, I output
> that, but something is exploding.
>
> Any ideas what it could be? As always, thanks.
>
> Here's from grunt:
>
> java.io.IOException: java.lang.RuntimeException: Unexpected data type -1
> found in stream.
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.runPipeline(PigMapReduce.java:438)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.processOnePackageOutput(PigMapReduce.java:401)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:381)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:251)
> at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
> at
> org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:566)
> at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:408)
> at
> org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:216)
> Caused by: java.lang.RuntimeException: Unexpected data type -1 found in
> stream.
> at
> org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:478)
> at
> org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
> at
> org.apache.pig.data.BinInterSedes.writeBag(BinInterSedes.java:522)
> at
> org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:361)
> at
> org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
> at
> org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:357)
> at
> org.apache.pig.data.BinInterSedes.writeTuple(BinInterSedes.java:541)
> at
> org.apache.pig.data.BinInterSedes.writeDatum(BinInterSedes.java:357)
> at
> org.apache.pig.impl.io.InterRecordWriter.write(InterRecordWriter.java:73)
> at org.apache.pig.impl.io.InterStorage.putNext(InterStorage.java:87)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat$PigRecordWriter.write(PigOutputFormat.java:138)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat$PigRecordWriter.write(PigOutputFormat.java:97)
> at
> org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:508)
> at
> org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
> at
> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.runPipeline(PigMapReduce.java:436)
> Here's from the logfile:Pig Stack Trace
> ---------------
> ERROR 1066: Unable to open iterator for alias E
>
> org.apache.pig.impl.logicalLayer.FrontendException: ERROR 1066: Unable to
> open iterator for alias E
> at org.apache.pig.PigServer.openIterator(PigServer.java:754)
> at
> org.apache.pig.tools.grunt.GruntParser.processDump(GruntParser.java:612)
> at
> org.apache.pig.tools.pigscript.parser.PigScriptParser.parse(PigScriptParser.java:303)
> at
> org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:165)
> at
> org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:141)
> at org.apache.pig.tools.grunt.Grunt.run(Grunt.java:76)
> at org.apache.pig.Main.run(Main.java:465)
> at org.apache.pig.Main.main(Main.java:107)
> Caused by: java.io.IOException: Job terminated with anomalous status FAILED
> at org.apache.pig.PigServer.openIterator(PigServer.java:744)
> ... 7 more
> ================================================================================
>