You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pr...@apache.org on 2009/10/27 02:13:22 UTC
svn commit: r830041 [2/3] - in /hadoop/pig/branches/load-store-redesign:
src/org/apache/pig/ src/org/apache/pig/backend/executionengine/
src/org/apache/pig/backend/hadoop/datastorage/
src/org/apache/pig/backend/hadoop/executionengine/ src/org/apache/pi...
Added: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigRecordReader.java?rev=830041&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigRecordReader.java (added)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigRecordReader.java Tue Oct 27 01:13:19 2009
@@ -0,0 +1,128 @@
+/**
+ *
+ */
+package org.apache.pig.backend.hadoop.executionengine.mapReduceLayer;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.pig.FuncSpec;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+
+/**
+ * A wrapper around the actual RecordReader and loadfunc - this is needed for
+ * two reasons
+ * 1) To intercept the initialize call from hadoop and initialize the underlying
+ * actual RecordReader with the right Context object - this is achieved by
+ * looking up the Context corresponding to the input split this Reader is
+ * supposed to process
+ * 2) We need to give hadoop consistent key-value types - text and tuple
+ * respectively - so PigRecordReader will call underlying Loader's getNext() to
+ * get the Tuple value - the key is null text since key is not used in input to
+ * map() in Pig.
+ */
+public class PigRecordReader extends RecordReader<Text, Tuple> {
+
+ /**
+ * the current Tuple value as returned by underlying
+ * {@link LoadFunc#getNext()}
+ */
+ Tuple curValue = null;
+
+ // the underlying RecordReader used by the loader
+ @SuppressWarnings("unchecked")
+ private RecordReader wrappedReader;
+
+ // the loader object
+ private LoadFunc loadfunc;
+
+ /**
+ * the Configuration object with data specific to the input the underlying
+ * RecordReader will process (this is obtained after a
+ * {@link LoadFunc#setLocation(String, org.apache.hadoop.mapreduce.Job)}
+ * call and hence can contain specific properties the underlying
+ * {@link InputFormat} might have put in.
+ */
+ private Configuration inputSpecificConf;
+ /**
+ * @param conf
+ *
+ */
+ public PigRecordReader(RecordReader wrappedReader,
+ LoadFunc loadFunc, Configuration conf) {
+ this.wrappedReader = wrappedReader;
+ this.loadfunc = loadFunc;
+ this.inputSpecificConf = conf;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#close()
+ */
+ @Override
+ public void close() throws IOException {
+ wrappedReader.close();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#getCurrentKey()
+ */
+ @Override
+ public Text getCurrentKey() throws IOException, InterruptedException {
+ // In pig we don't really use the key in the input to the map - so send
+ // null
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#getCurrentValue()
+ */
+ @Override
+ public Tuple getCurrentValue() throws IOException, InterruptedException {
+ return curValue;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#getProgress()
+ */
+ @Override
+ public float getProgress() throws IOException, InterruptedException {
+ return wrappedReader.getProgress();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#initialize(org.apache.hadoop.mapreduce.InputSplit, org.apache.hadoop.mapreduce.TaskAttemptContext)
+ */
+ @Override
+ public void initialize(InputSplit split, TaskAttemptContext context)
+ throws IOException, InterruptedException {
+ // initialize the underlying actual RecordReader with the right Context
+ // object - this is achieved by looking up the Context corresponding to
+ // the input split this Reader is supposed to process
+ PigSplit pigSplit = (PigSplit)split;
+ ConfigurationUtil.mergeConf(context.getConfiguration(),
+ inputSpecificConf);
+
+ // now invoke initialize() on underlying RecordReader with
+ // the "adjusted" conf
+ wrappedReader.initialize(pigSplit.getWrappedSplit(), context);
+ loadfunc.prepareToRead(wrappedReader, pigSplit);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.RecordReader#nextKeyValue()
+ */
+ @Override
+ public boolean nextKeyValue() throws IOException, InterruptedException {
+ curValue = loadfunc.getNext();
+ return curValue != null;
+ }
+
+}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSplit.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSplit.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSplit.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigSplit.java Tue Oct 27 01:13:19 2009
@@ -22,140 +22,139 @@
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.io.OutputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
-import org.apache.hadoop.fs.BlockLocation;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.mapred.InputSplit;
-import org.apache.pig.FuncSpec;
-import org.apache.pig.LoadFunc;
-import org.apache.pig.builtin.PigStorage;
-import org.apache.pig.impl.PigContext;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.SerializationFactory;
+import org.apache.hadoop.io.serializer.Serializer;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.util.ReflectionUtils;
import org.apache.pig.impl.plan.OperatorKey;
/**
* The main split class that maintains important
* information about the input split.
*
+ * The reason this class implements Configurable is so that Hadoop will call
+ * {@link Configurable#setConf(Configuration)} on the backend so we can use
+ * the Configuration to create the SerializationFactory to deserialize the
+ * wrapped InputSplit.
*/
-public class PigSplit implements InputSplit {
- //The input file this split represents
- Path file;
-
- //The start pos from which this split starts
- //in file
- long start;
-
- //The length of this split
- long length;
-
+public class PigSplit extends InputSplit implements Writable, Configurable {
//The operators to which the tuples from this
//input file are attached. These are the successors
//of the load operator representing this input
- ArrayList<OperatorKey> targetOps;
-
- //The load function that parses this split and
- //produces tuples from it
- String parser;
-
- //The filesystem on which file resides
- FileSystem fs;
-
- //The context in which Pig was running when this
- //job got submitted.
- PigContext pigContext;
-
- public PigSplit() {
- }
-
- @SuppressWarnings("unchecked")
- public PigSplit(PigContext pigContext, FileSystem fs, Path path,
- String parser, List<OperatorKey> targetOps, long start, long length) {
- this.fs = fs;
- this.file = path;
- this.start = start;
- this.length = length;
- this.targetOps = (ArrayList) targetOps;
- this.parser = parser;
- this.pigContext = pigContext;
- }
-
- public String getParser() {
- return parser;
- }
+ private ArrayList<OperatorKey> targetOps;
- public long getStart() {
- return start;
- }
+ // index starting from 0 representing the input number
+ // So if we have 3 inputs (say for a 3 way join), then the
+ // splits corresponding to the first input will have an index of 0, those
+ // corresponding to the second will have an index of 1 and so on
+ // This will be used to get the input specific Configuration for use
+ // by PigInputFormat while creating and initializing the RecordReader
+ private int inputIndex;
+
+ // The real InputSplit this split is wrapping
+ private InputSplit wrappedSplit;
- public long getLength() {
- return length;
+ // index of the wrappedSplit in the list of splits returned by
+ // InputFormat.getSplits()
+ // This will be used by MergeJoinIndexer to record the split # in the
+ // index
+ private int splitIndex;
+
+ /**
+ * the job Configuration
+ */
+ private Configuration conf;
+
+ /**
+ * total number of splits - required by skew join
+ */
+ private int totalSplits;
+
+ // this seems necessary for Hadoop to instatiate this split on the
+ // backend
+ public PigSplit() {}
+
+ public PigSplit(InputSplit wrappedSplit, int inputIndex,
+ List<OperatorKey> targetOps, int splitIndex) {
+ this.wrappedSplit = wrappedSplit;
+ this.inputIndex = inputIndex;
+ this.targetOps = new ArrayList<OperatorKey>(targetOps);
+ this.splitIndex = splitIndex;
}
- public Path getPath() {
- return file;
- }
public List<OperatorKey> getTargetOps() {
- return targetOps;
+ return new ArrayList<OperatorKey>(targetOps);
}
+
- public LoadFunc getLoadFunction() {
- LoadFunc loader = null;
- if (this.parser == null) {
- loader = new PigStorage();
- } else {
- try {
- loader = (LoadFunc) PigContext
- .instantiateFuncFromSpec(new FuncSpec(this.parser));
- } catch (Exception exp) {
- throw new RuntimeException("can't instantiate " + parser);
- }
- }
- return loader;
+ /**
+ * This methods returns the actual InputSplit (as returned by the
+ * {@link InputFormat}) which this class is wrapping.
+ * @return the wrappedSplit
+ */
+ public InputSplit getWrappedSplit() {
+ return wrappedSplit;
}
-
- public String[] getLocations() throws IOException {
- FileStatus status = fs.getFileStatus(file);
- BlockLocation[] b = fs.getFileBlockLocations(status, start, length);
- int total = 0;
- for (int i = 0; i < b.length; i++) {
- total += b[i].getHosts().length;
- }
- String locations[] = new String[total];
- int count = 0;
- for (int i = 0; i < b.length; i++) {
- String hosts[] = b[i].getHosts();
- for (int j = 0; j < hosts.length; j++) {
- locations[count++] = hosts[j];
- }
- }
- return locations;
+
+ @Override
+ public String[] getLocations() throws IOException, InterruptedException {
+ return wrappedSplit.getLocations();
}
- @SuppressWarnings("unchecked")
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.mapreduce.InputSplit#getLength()
+ */
+ @Override
+ public long getLength() throws IOException, InterruptedException {
+ return wrappedSplit.getLength();
+ }
+
public void readFields(DataInput is) throws IOException {
- file = new Path(is.readUTF());
- start = is.readLong();
- length = is.readLong();
- pigContext = (PigContext) readObject(is);
- targetOps = (ArrayList) readObject(is);
- parser = is.readUTF();
+ splitIndex = is.readInt();
+ inputIndex = is.readInt();
+ targetOps = (ArrayList<OperatorKey>) readObject(is);
+ String splitClassName = is.readUTF();
+ try {
+ Class splitClass = conf.getClassByName(splitClassName);
+ wrappedSplit = (InputSplit)
+ ReflectionUtils.newInstance(splitClass, conf);
+ SerializationFactory sf = new SerializationFactory(conf);
+ Deserializer d = sf.getDeserializer(splitClass);
+ d.open((InputStream) is);
+ d.deserialize(wrappedSplit);
+ d.close();
+ } catch (ClassNotFoundException e) {
+ throw new IOException(e);
+ }
+
}
public void write(DataOutput os) throws IOException {
- os.writeUTF(file.toString());
- os.writeLong(start);
- os.writeLong(length);
- writeObject(pigContext, os);
+ os.writeInt(splitIndex);
+ os.writeInt(inputIndex);
writeObject(targetOps, os);
- os.writeUTF(parser);
+ os.writeUTF(wrappedSplit.getClass().getName());
+ SerializationFactory sf = new SerializationFactory(conf);
+ Serializer s =
+ sf.getSerializer(wrappedSplit.getClass());
+ s.open((OutputStream) os);
+ s.serialize(wrappedSplit);
+ s.close();
+
}
private void writeObject(Serializable obj, DataOutput os)
@@ -182,4 +181,44 @@
}
}
+ public int getSplitIndex() {
+ return splitIndex;
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.conf.Configurable#getConf()
+ */
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.hadoop.conf.Configurable#setConf(org.apache.hadoop.conf.Configuration)
+ */
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ }
+
+ public int getInputIndex() {
+ return inputIndex;
+ }
+
+ /**
+ * @return the totalSplits
+ */
+ public int getTotalSplits() {
+ return totalSplits;
+ }
+
+ /**
+ * @param totalSplits the totalSplits to set
+ */
+ public void setTotalSplits(int totalSplits) {
+ this.totalSplits = totalSplits;
+ }
+
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/partitioners/WeightedRangePartitioner.java Tue Oct 27 01:13:19 2009
@@ -40,6 +40,7 @@
import org.apache.pig.impl.builtin.FindQuantiles;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.io.ReadToEndLoader;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.io.NullableBytesWritable;
import org.apache.pig.impl.io.NullableDoubleWritable;
@@ -93,9 +94,9 @@
try{
InputStream is = FileLocalizer.openDFSFile(quantilesFile,ConfigurationUtil.toProperties(job));
- BinStorage loader = new BinStorage();
+ ReadToEndLoader loader = new ReadToEndLoader(new BinStorage(), job,
+ quantilesFile, 0);
DataBag quantilesList;
- loader.bindTo(quantilesFile, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
Tuple t = loader.getNext();
if(t!=null)
{
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java Tue Oct 27 01:13:19 2009
@@ -24,6 +24,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.FuncSpec;
+import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
@@ -49,6 +50,7 @@
public class POCast extends ExpressionOperator {
private FuncSpec loadFSpec = null;
transient private LoadFunc load;
+ transient private LoadCaster caster;
transient private Log log = LogFactory.getLog(getClass());
private boolean castNotNeeded = false;
private Byte realType = null;
@@ -72,6 +74,7 @@
this.load = (LoadFunc) PigContext
.instantiateFuncFromSpec(this.loadFSpec);
}
+ this.caster = load.getLoadCaster();
}
public void setLoadFSpec(FuncSpec lf) {
@@ -135,8 +138,8 @@
return res;
}
try {
- if (null != load) {
- res.result = load.bytesToInteger(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToInteger(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to int.";
@@ -262,8 +265,8 @@
return res;
}
try {
- if (null != load) {
- res.result = load.bytesToLong(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToLong(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to long.";
@@ -384,8 +387,8 @@
return res;
}
try {
- if (null != load) {
- res.result = load.bytesToDouble(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToDouble(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to double.";
@@ -505,8 +508,8 @@
return res;
}
try {
- if (null != load) {
- res.result = load.bytesToFloat(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToFloat(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to float.";
@@ -628,8 +631,8 @@
return res;
}
try {
- if (null != load) {
- res.result = load.bytesToCharArray(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToCharArray(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to string.";
@@ -650,9 +653,9 @@
Result res = in.getNext(b);
if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
if (((Boolean) res.result) == true)
- res.result = "1";
+ res.result = new String("1");
else
- res.result = "0";
+ res.result = new String("1");
}
return res;
}
@@ -660,7 +663,7 @@
Integer dummyI = null;
Result res = in.getNext(dummyI);
if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
- res.result = ((Integer) res.result).toString();
+ res.result = new String(((Integer) res.result).toString());
}
return res;
}
@@ -670,7 +673,7 @@
Result res = in.getNext(d);
if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
// res.result = DataType.toInteger(res.result);
- res.result = ((Double) res.result).toString();
+ res.result = new String(((Double) res.result).toString());
}
return res;
}
@@ -680,7 +683,7 @@
Long l = null;
Result res = in.getNext(l);
if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
- res.result = ((Long) res.result).toString();
+ res.result = new String(((Long) res.result).toString());
}
return res;
}
@@ -689,7 +692,7 @@
Float f = null;
Result res = in.getNext(f);
if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
- res.result = ((Float) res.result).toString();
+ res.result = new String(((Float) res.result).toString());
}
return res;
}
@@ -753,8 +756,8 @@
}
try {
- if (null != load) {
- res.result = load.bytesToTuple(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToTuple(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to tuple.";
@@ -841,8 +844,8 @@
}
try {
- if (null != load) {
- res.result = load.bytesToBag(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToBag(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to bag.";
@@ -929,8 +932,8 @@
}
try {
- if (null != load) {
- res.result = load.bytesToMap(dba.get());
+ if (null != caster) {
+ res.result = caster.bytesToMap(dba.get());
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to map.";
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLoad.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLoad.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLoad.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLoad.java Tue Oct 27 01:13:19 2009
@@ -31,7 +31,9 @@
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.impl.io.ReadToEndLoader;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
@@ -98,11 +100,21 @@
*/
public void setUp() throws IOException{
String filename = lFile.getFileName();
- loader = (LoadFunc)PigContext.instantiateFuncFromSpec(lFile.getFuncSpec());
+ LoadFunc origloader =
+ (LoadFunc)PigContext.instantiateFuncFromSpec(lFile.getFuncSpec());
+ loader = new ReadToEndLoader(origloader,
+ ConfigurationUtil.toConfiguration(pc.getProperties()),
+ filename,
+ 0);
+
+ // XXX : FIXME need to get this to work with new loadfunc interface - the
+ // below code is for merge join - hopefully we will no longer need it
+ // and then we can just get rid of it and the rest should be fine.
is = (this.offset == 0) ? FileLocalizer.open(filename, pc) : FileLocalizer.open(filename, this.offset,pc);
- loader.bindTo(filename , new BufferedPositionedInputStream(is), this.offset, Long.MAX_VALUE);
+// loader.bindTo(filename , new BufferedPositionedInputStream(is), this.offset, Long.MAX_VALUE);
+
}
/**
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java Tue Oct 27 01:13:19 2009
@@ -194,4 +194,10 @@
public Schema getSchema() {
return schema;
}
+
+ public StoreFunc getStoreFunc() {
+ return (StoreFunc)
+ PigContext.instantiateFuncFromSpec(sFile.getFuncSpec());
+ }
+
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/util/PlanHelper.java Tue Oct 27 01:13:19 2009
@@ -17,6 +17,7 @@
*/
package org.apache.pig.backend.hadoop.executionengine.physicalLayer.util;
+import java.util.ArrayList;
import java.util.List;
import java.util.LinkedList;
@@ -48,7 +49,7 @@
* @param plan
* @return List of stores (could be empty)
*/
- public static List<POStore> getStores(PhysicalPlan plan) throws VisitorException {
+ public static LinkedList<POStore> getStores(PhysicalPlan plan) throws VisitorException {
LoadStoreFinder finder = new LoadStoreFinder(plan);
finder.visit();
@@ -60,7 +61,7 @@
* @param plan
* @return List of loads (could be empty)
*/
- public static List<POLoad> getLoads(PhysicalPlan plan) throws VisitorException {
+ public static LinkedList<POLoad> getLoads(PhysicalPlan plan) throws VisitorException {
LoadStoreFinder finder = new LoadStoreFinder(plan);
finder.visit();
@@ -85,8 +86,8 @@
}
private static class LoadStoreFinder extends PhyPlanVisitor {
- private List<POLoad> loads;
- private List<POStore> stores;
+ private LinkedList<POLoad> loads;
+ private LinkedList<POStore> stores;
LoadStoreFinder(PhysicalPlan plan) {
super(plan, new DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(plan));
@@ -111,11 +112,11 @@
loads.add(load);
}
- public List<POStore> getStores() {
+ public LinkedList<POStore> getStores() {
return stores;
}
- public List<POLoad> getLoads() {
+ public LinkedList<POLoad> getLoads() {
return loads;
}
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java Tue Oct 27 01:13:19 2009
@@ -20,6 +20,7 @@
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler;
import org.apache.pig.builtin.BinStorage;
import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.impl.io.ReadToEndLoader;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.DefaultTupleFactory;
@@ -100,9 +101,9 @@
} else {
is = FileLocalizer.openDFSFile(keyDistFile);
}
- BinStorage loader = new BinStorage();
+ ReadToEndLoader loader = new ReadToEndLoader(new BinStorage(), job,
+ keyDistFile, 0);
DataBag partitionList;
- loader.bindTo(keyDistFile, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
Tuple t = loader.getNext();
if(t==null) {
throw new RuntimeException("Empty samples file");
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/hbase/HBaseStorage.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/hbase/HBaseStorage.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/hbase/HBaseStorage.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/hadoop/hbase/HBaseStorage.java Tue Oct 27 01:13:19 2009
@@ -20,15 +20,22 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.pig.ExecType;
+import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.Slice;
import org.apache.pig.Slicer;
import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.builtin.Utf8StorageConverter;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
@@ -146,27 +153,55 @@
// It is just a mock class to let the UDF be casted to a LOADFUNC during
// parsing.
+
@Override
- public void bindTo(String fileName, BufferedPositionedInputStream is,
- long offset, long end) throws IOException {
+ public Tuple getNext() throws IOException {
// do nothing
+ return null;
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
@Override
- public Schema determineSchema(String fileName, ExecType execType,
- DataStorage storage) throws IOException {
- // do nothing
- return null;
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
@Override
- public void fieldsToRead(Schema schema) {
- // do nothing
+ public InputFormat getInputFormat() {
+ // TODO Auto-generated method stub
+ return null;
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
@Override
- public Tuple getNext() throws IOException {
- // do nothing
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
return null;
}
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalJob.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalJob.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalJob.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalJob.java Tue Oct 27 01:13:19 2009
@@ -18,14 +18,23 @@
package org.apache.pig.backend.local.executionengine;
+import java.io.IOException;
import java.io.OutputStream;
import java.io.InputStream;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.executionengine.ExecJob;
import org.apache.pig.data.Tuple;
@@ -34,6 +43,7 @@
import org.apache.pig.LoadFunc;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.io.ReadToEndLoader;
import org.apache.pig.tools.pigstats.PigStats;
@@ -74,14 +84,18 @@
public Iterator<Tuple> getResults() throws ExecException {
final LoadFunc p;
+ final InputFormat inputFormat;
try{
- p = (LoadFunc)PigContext.instantiateFuncFromSpec(outFileSpec.getFuncSpec());
-
- InputStream is = FileLocalizer.open(outFileSpec.getFileName(), pigContext);
-
- p.bindTo(outFileSpec.getFileName(), new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
-
+ LoadFunc origLoadFunc = (LoadFunc)PigContext.instantiateFuncFromSpec(outFileSpec.getFuncSpec());
+
+ String fileName = outFileSpec.getFileName();
+ if(!fileName.startsWith("file://")) {
+ fileName = "file://" + fileName;
+ }
+ //XXX: FIXME: ensure this works in local mode (part of load-store redesign changes)
+ p = new ReadToEndLoader(origLoadFunc, ConfigurationUtil.toConfiguration(
+ pigContext.getProperties()), fileName, 0);
}catch (Exception e){
throw new ExecException("Unable to get results for " + outFileSpec, e);
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalPOStoreImpl.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalPOStoreImpl.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalPOStoreImpl.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/backend/local/executionengine/LocalPOStoreImpl.java Tue Oct 27 01:13:19 2009
@@ -32,6 +32,7 @@
* execution engine. It creates and manages the store function and the
* output stream of the store.
*/
+// XXX: FIXME make this work with new loadstore redesign
public class LocalPOStoreImpl extends POStoreImpl {
private OutputStream os;
@@ -49,13 +50,13 @@
this.sFile = sFile;
storer = (StoreFunc)PigContext.instantiateFuncFromSpec(sFile.getFuncSpec());
os = FileLocalizer.create(sFile.getFileName(), pc);
- storer.bindTo(os);
+// storer.bindTo(os);
return storer;
}
@Override
public void tearDown() throws IOException{
- storer.finish();
+// storer.finish();
os.close();
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinStorage.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinStorage.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinStorage.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinStorage.java Tue Oct 27 01:13:19 2009
@@ -23,108 +23,59 @@
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.Map;
-import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.Log;
-import org.apache.pig.ExecType;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.pig.LoadCaster;
import org.apache.pig.PigException;
import org.apache.pig.PigWarning;
+import org.apache.pig.ResourceSchema;
import org.apache.pig.ReversibleLoadStoreFunc;
-import org.apache.pig.SamplableLoader;
-import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataReaderWriter;
-import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
-import org.apache.pig.impl.io.BufferedPositionedInputStream;
-import org.apache.pig.impl.io.FileLocalizer;
-import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.io.BinStorageInputFormat;
+import org.apache.pig.impl.io.BinStorageRecordReader;
import org.apache.pig.impl.util.LogUtils;
-public class BinStorage implements ReversibleLoadStoreFunc, SamplableLoader {
+public class BinStorage implements ReversibleLoadStoreFunc, LoadCaster {
public static final int RECORD_1 = 0x01;
public static final int RECORD_2 = 0x02;
public static final int RECORD_3 = 0x03;
- protected BufferedPositionedInputStream in = null;
+ Iterator<Tuple> i = null;
private static final Log mLog = LogFactory.getLog(BinStorage.class);
- private DataInputStream inData = null;
protected long end = Long.MAX_VALUE;
+ private BinStorageRecordReader recReader = null;
+
/**
* Simple binary nested reader format
*/
public BinStorage() {
}
- @Override
- public long getPosition() throws IOException {
- return in.getPosition();
- }
-
- @Override
- public long skip(long n) throws IOException {
- return in.skip(n);
- }
-
public Tuple getNext() throws IOException {
-
- int b = 0;
-// skip to next record
- while (true) {
- if (in == null || in.getPosition() >=end) {
- return null;
- }
- // check if we saw RECORD_1 in our last attempt
- // this can happen if we have the following
- // sequence RECORD_1-RECORD_1-RECORD_2-RECORD_3
- // After reading the second RECORD_1 in the above
- // sequence, we should not look for RECORD_1 again
- if(b != RECORD_1) {
- b = in.read();
- if(b != RECORD_1 && b != -1) {
- continue;
- }
- if(b == -1) return null;
- }
- b = in.read();
- if(b != RECORD_2 && b != -1) {
- continue;
- }
- if(b == -1) return null;
- b = in.read();
- if(b != RECORD_3 && b != -1) {
- continue;
- }
- if(b == -1) return null;
- b = in.read();
- if(b != DataType.TUPLE && b != -1) {
- continue;
- }
- if(b == -1) return null;
- break;
- }
- try {
- // if we got here, we have seen RECORD_1-RECORD_2-RECORD_3-TUPLE_MARKER
- // sequence - lets now read the contents of the tuple
- return (Tuple)DataReaderWriter.readDatum(inData, DataType.TUPLE);
- } catch (ExecException ee) {
- throw ee;
+ if(recReader.nextKeyValue()) {
+ return recReader.getCurrentValue();
+ } else {
+ return null;
}
}
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
- this.in = in;
- inData = new DataInputStream(in);
- this.end = end;
- }
-
-
DataOutputStream out = null;
public void bindTo(OutputStream os) throws IOException {
@@ -254,50 +205,6 @@
}
}
- /* (non-Javadoc)
- * @see org.apache.pig.LoadFunc#determineSchema(java.lang.String, org.apache.pig.ExecType, org.apache.pig.backend.datastorage.DataStorage)
- */
- public Schema determineSchema(String fileName, ExecType execType,
- DataStorage storage) throws IOException {
-
- if (!FileLocalizer.fileExists(fileName, storage)) {
- // At compile time in batch mode, the file may not exist
- // (such as intermediate file). Just return null - the
- // same way as we would if we did not get a valid record
- return null;
- }
-
- InputStream is = FileLocalizer.open(fileName, execType, storage);
-
- bindTo(fileName, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
- // get the first record from the input file
- // and figure out the schema from the data in
- // the first record
- Tuple t = getNext();
- is.close();
- if(t == null) {
- // we couldn't get a valid record from the input
- return null;
- }
- int numFields = t.size();
- Schema s = new Schema();
- for (int i = 0; i < numFields; i++) {
- try {
- s.add(DataType.determineFieldSchema(t.get(i)));
- } catch (Exception e) {
- int errCode = 2104;
- String msg = "Error while determining schema of BinStorage data.";
- throw new ExecException(msg, errCode, PigException.BUG, e);
- }
- }
- return s;
- }
-
- public void fieldsToRead(Schema schema) {
- // TODO Auto-generated method stub
-
- }
-
public byte[] toBytes(DataBag bag) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
@@ -401,21 +308,97 @@
}
return baos.toByteArray();
}
- public boolean equals(Object obj) {
- return true;
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // nothing to be done for now
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
+ @Override
+ public InputFormat getInputFormat() {
+ return new BinStorageInputFormat();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ return this;
}
/* (non-Javadoc)
- * @see org.apache.pig.StoreFunc#getStorePreparationClass()
*/
@Override
- public Class getStorePreparationClass() throws IOException {
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ recReader = (BinStorageRecordReader)reader;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ FileInputFormat.setInputPaths(job, location);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#allFinished(org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void allFinished(Job job) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#doneWriting()
+ */
+ @Override
+ public void doneWriting() {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#getOutputFormat()
+ */
+ @Override
+ public OutputFormat getOutputFormat() {
// TODO Auto-generated method stub
return null;
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#prepareToWrite(org.apache.hadoop.mapreduce.RecordWriter)
+ */
@Override
- public Tuple getSampledTuple() throws IOException {
- return this.getNext();
+ public void prepareToWrite(RecordWriter writer) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setSchema(org.apache.pig.ResourceSchema)
+ */
+ @Override
+ public void setSchema(ResourceSchema s) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setStoreLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
}
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinaryStorage.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinaryStorage.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinaryStorage.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/BinaryStorage.java Tue Oct 27 01:13:19 2009
@@ -22,11 +22,21 @@
import java.io.OutputStream;
import java.net.URL;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.pig.ExecType;
+import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
+import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DefaultTupleFactory;
@@ -45,7 +55,8 @@
* are to be sent in-whole for processing without any splitting and
* interpretation of their data.
*/
-public class BinaryStorage extends Utf8StorageConverter implements LoadFunc, StoreFunc {
+// XXX : FIXME - make this work with new load-store redesign
+public class BinaryStorage implements LoadFunc, StoreFunc {
// LoadFunc
private static final int DEFAULT_BUFFER_SIZE = 64*1024;
protected int bufferSize = DEFAULT_BUFFER_SIZE;
@@ -73,13 +84,6 @@
this.bufferSize = bufferSize;
}
- public void bindTo(String fileName, BufferedPositionedInputStream in,
- long offset, long end) throws IOException {
- this.in = in;
- this.offset = offset;
- this.end = end;
- }
-
public Tuple getNext() throws IOException {
// Sanity check
if (in == null || in.getPosition() > end) {
@@ -144,28 +148,101 @@
}
/* (non-Javadoc)
- * @see org.apache.pig.LoadFunc#determineSchema(java.lang.String, org.apache.pig.ExecType, org.apache.pig.backend.datastorage.DataStorage)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
*/
- public Schema determineSchema(String fileName, ExecType execType,
- DataStorage storage) throws IOException {
+ @Override
+ public InputFormat getInputFormat() {
// TODO Auto-generated method stub
return null;
}
-
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#allFinished(org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void allFinished(Job job) {
+ // TODO Auto-generated method stub
+
+ }
+
/* (non-Javadoc)
- * @see org.apache.pig.LoadFunc#fieldsToRead(org.apache.pig.impl.logicalLayer.schema.Schema)
+ * @see org.apache.pig.StoreFunc#doneWriting()
*/
- public void fieldsToRead(Schema schema) {
+ @Override
+ public void doneWriting() {
// TODO Auto-generated method stub
}
/* (non-Javadoc)
- * @see org.apache.pig.StoreFunc#getStorePreparationClass()
+ * @see org.apache.pig.StoreFunc#getOutputFormat()
*/
@Override
- public Class getStorePreparationClass() throws IOException {
+ public OutputFormat getOutputFormat() {
// TODO Auto-generated method stub
return null;
}
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#prepareToWrite(org.apache.hadoop.mapreduce.RecordWriter)
+ */
+ @Override
+ public void prepareToWrite(RecordWriter writer) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setSchema(org.apache.pig.ResourceSchema)
+ */
+ @Override
+ public void setSchema(ResourceSchema s) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setStoreLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigDump.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigDump.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigDump.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigDump.java Tue Oct 27 01:13:19 2009
@@ -20,11 +20,16 @@
import java.io.IOException;
import java.io.OutputStream;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.TupleFormat;
+// XXX: FIXME - make this work with new load-store redesign
public class PigDump implements StoreFunc {
public static String recordDelimiter = "\n";
@@ -45,12 +50,57 @@
}
/* (non-Javadoc)
- * @see org.apache.pig.StoreFunc#getStorePreparationClass()
+ * @see org.apache.pig.StoreFunc#allFinished(org.apache.hadoop.mapreduce.Job)
*/
@Override
- public Class getStorePreparationClass() throws IOException {
+ public void allFinished(Job job) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#doneWriting()
+ */
+ @Override
+ public void doneWriting() {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#getOutputFormat()
+ */
+ @Override
+ public OutputFormat getOutputFormat() {
// TODO Auto-generated method stub
return null;
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#prepareToWrite(org.apache.hadoop.mapreduce.RecordWriter)
+ */
+ @Override
+ public void prepareToWrite(RecordWriter writer) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setSchema(org.apache.pig.ResourceSchema)
+ */
+ @Override
+ public void setSchema(ResourceSchema s) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setStoreLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigStorage.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigStorage.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigStorage.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/PigStorage.java Tue Oct 27 01:13:19 2009
@@ -17,46 +17,67 @@
*/
package org.apache.pig.builtin;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
-import java.util.Map;
import java.util.Iterator;
+import java.util.Map;
-import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
-import org.apache.pig.ExecType;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.pig.LoadCaster;
import org.apache.pig.PigException;
-import org.apache.pig.SamplableLoader;
+import org.apache.pig.ResourceSchema;
import org.apache.pig.ReversibleLoadStoreFunc;
-import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.DataByteArray;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
-import org.apache.pig.impl.io.BufferedPositionedInputStream;
-import org.apache.pig.impl.io.PigLineRecordReader;
-import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.data.TupleFactory;
/**
* A load function that parses a line of input into fields using a delimiter to set the fields. The
* delimiter is given as a regular expression. See String.split(delimiter) and
* http://java.sun.com/j2se/1.5.0/docs/api/java/util/regex/Pattern.html for more information.
*/
-public class PigStorage extends Utf8StorageConverter
- implements ReversibleLoadStoreFunc, SamplableLoader {
- protected PigLineRecordReader in = null;
+public class PigStorage
+ implements ReversibleLoadStoreFunc {
+ protected RecordReader in = null;
+ protected RecordWriter writer = null;
protected final Log mLog = LogFactory.getLog(getClass());
long end = Long.MAX_VALUE;
private byte recordDel = '\n';
private byte fieldDel = '\t';
private ArrayList<Object> mProtoTuple = null;
+ private int os;
+ private TupleFactory mTupleFactory = TupleFactory.getInstance();
+ private static final int OS_UNIX = 0;
+ private static final int OS_WINDOWS = 1;
private static final String UTF8 = "UTF-8";
-
- public PigStorage() {}
+ private static final int BUFFER_SIZE = 1024;
+
+ public PigStorage() {
+ os = OS_UNIX;
+ if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
+ os = OS_WINDOWS;
+ }
/**
* Constructs a Pig loader that uses specified regex as a field delimiter.
@@ -89,81 +110,40 @@
}
}
- public long getPosition() throws IOException {
- return in.getPosition();
- }
-
- public long skip(long n) throws IOException {
-
- Text t = new Text();
- long sofar = 0;
- while (sofar < n) {
- /*
- * By calling next we skip more than required bytes
- * but this skip will be only until end of record.. aka line
- */
- if (in.next( t)) {
- sofar += t.getLength() + 1;
- } else {
- // End of file
- return sofar;
- }
- }
- return sofar;
- }
-
public Tuple getNext() throws IOException {
- if (in == null || in.getPosition() > end) {
- return null;
- }
- Text value = new Text();
- boolean notDone = in.next(value);
- if (!notDone) {
- return null;
- }
-
- byte[] buf = value.getBytes();
- int len = value.getLength();
- int start = 0;
- for (int i = 0; i < len; i++) {
- if (buf[i] == fieldDel) {
- readField(buf, start, i);
- start = i + 1;
+ try {
+ boolean notDone = in.nextKeyValue();
+ if (!notDone) {
+ return null;
+ }
+ Text value = (Text) in.getCurrentValue();
+ byte[] buf = value.getBytes();
+ int len = value.getLength();
+ int start = 0;
+ for (int i = 0; i < len; i++) {
+ if (buf[i] == fieldDel) {
+ readField(buf, start, i);
+ start = i + 1;
+ }
}
+ // pick up the last field
+ if (start <= len) {
+ readField(buf, start, len);
+ }
+ Tuple t = mTupleFactory.newTupleNoCopy(mProtoTuple);
+ // System.out.println( "Arity:" + t.size() + " Value: " + value.toString() );
+ mProtoTuple = null;
+ return t;
+ } catch (InterruptedException e) {
+ // FIXME: XXX - include errorcode
+ throw new IOException("Error getting input");
}
- // pick up the last field
- if (start <= len) {
- readField(buf, start, len);
- }
- Tuple t = mTupleFactory.newTupleNoCopy(mProtoTuple);
- // System.out.println( "Arity:" + t.size() + " Value: " + value.toString() );
- mProtoTuple = null;
- return t;
}
- public Tuple getSampledTuple() throws IOException {
- return getNext();
- }
+ ByteArrayOutputStream mOut = new ByteArrayOutputStream(BUFFER_SIZE);
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
- this.in = new PigLineRecordReader( in, offset, end );
- this.end = end;
-
- // Since we are not block aligned we throw away the first
- // record and cound on a different instance to read it
- if (offset != 0) {
- getNext();
- }
- }
-
- OutputStream mOut;
- public void bindTo(OutputStream os) throws IOException {
- mOut = os;
- }
-
- @SuppressWarnings("unchecked")
private void putField(Object field) throws IOException {
//string constants for each delimiter
String tupleBeginDelim = "(";
@@ -285,13 +265,17 @@
putField(field);
- if (i == sz - 1) {
- // last field in tuple.
- mOut.write(recordDel);
- } else {
+ if (i != sz - 1) {
mOut.write(fieldDel);
}
}
+ Text text = new Text(mOut.toByteArray());
+ try {
+ writer.write(null, text);
+ mOut.reset();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
}
public void finish() throws IOException {
@@ -310,33 +294,109 @@
}
}
+ public boolean equals(Object obj) {
+ return equals((PigStorage)obj);
+ }
+
+ public boolean equals(PigStorage other) {
+ return this.fieldDel == other.fieldDel;
+ }
+
/* (non-Javadoc)
- * @see org.apache.pig.LoadFunc#determineSchema(java.lang.String, org.apache.pig.ExecType, org.apache.pig.backend.datastorage.DataStorage)
+ * @see org.apache.pig.LoadFunc#doneReading()
*/
- public Schema determineSchema(String fileName, ExecType execType,
- DataStorage storage) throws IOException {
+ @Override
+ public void doneReading() {
// TODO Auto-generated method stub
- return null;
+
}
- public void fieldsToRead(Schema schema) {
- // do nothing
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
+ @Override
+ public InputFormat getInputFormat() {
+ // TODO Auto-generated method stub
+ return new TextInputFormat();
}
-
- public boolean equals(Object obj) {
- return equals((PigStorage)obj);
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return new Utf8StorageConverter();
}
- public boolean equals(PigStorage other) {
- return this.fieldDel == other.fieldDel;
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ in = reader;
}
/* (non-Javadoc)
- * @see org.apache.pig.StoreFunc#getStorePreparationClass()
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.conf.Configuration)
*/
-
- public Class getStorePreparationClass() throws IOException {
+ @Override
+ public void setLocation(String location, Job job)
+ throws IOException {
+ FileInputFormat.setInputPaths(job, location);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#allFinished(org.apache.hadoop.conf.Configuration)
+ */
+ @Override
+ public void allFinished(Job job) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#doneWriting()
+ */
+ @Override
+ public void doneWriting() {
// TODO Auto-generated method stub
- return null;
+
}
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#getOutputFormat()
+ */
+ @Override
+ public OutputFormat getOutputFormat() {
+ return new TextOutputFormat<Text, Text>();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#prepareToWrite(org.apache.hadoop.mapreduce.RecordWriter)
+ */
+ @Override
+ public void prepareToWrite(RecordWriter writer) {
+ this.writer = writer;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setLocation(java.lang.String)
+ */
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ job.getConfiguration().set("mapred.textoutputformat.separator", "");
+ FileOutputFormat.setOutputPath(job, new Path(location));
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.StoreFunc#setSchema(org.apache.pig.ResourceSchema)
+ */
+ @Override
+ public void setSchema(ResourceSchema s) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/TextLoader.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/TextLoader.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/TextLoader.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/TextLoader.java Tue Oct 27 01:13:19 2009
@@ -24,11 +24,18 @@
import java.nio.charset.Charset;
import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.pig.ExecType;
+import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
@@ -41,21 +48,13 @@
* This load function simply creates a tuple for each line of text that has a single field that
* contains the line of text.
*/
+//XXX : FIXME - make this work with new load-store redesign
public class TextLoader implements LoadFunc{
BufferedPositionedInputStream in;
final private static Charset utf8 = Charset.forName("UTF8");
long end;
private TupleFactory mTupleFactory = TupleFactory.getInstance();
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
- this.in = in;
- this.end = end;
- // Since we are not block aligned we throw away the first
- // record and count on a different instance to read it
- if (offset != 0)
- getNext();
- }
-
public Tuple getNext() throws IOException {
if (in == null || in.getPosition() > end)
return null;
@@ -217,4 +216,49 @@
String msg = "TextLoader does not support conversion from Tuple.";
throw new ExecException(msg, errCode, PigException.BUG);
}
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
+ @Override
+ public InputFormat getInputFormat() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/Utf8StorageConverter.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/Utf8StorageConverter.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/Utf8StorageConverter.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/builtin/Utf8StorageConverter.java Tue Oct 27 01:13:19 2009
@@ -25,6 +25,7 @@
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.Log;
+import org.apache.pig.LoadCaster;
import org.apache.pig.PigException;
import org.apache.pig.PigWarning;
import org.apache.pig.backend.executionengine.ExecException;
@@ -44,7 +45,7 @@
* and pig data types. It is intended to be extended by load and store
* functions (such as PigStorage).
*/
-abstract public class Utf8StorageConverter {
+public class Utf8StorageConverter implements LoadCaster {
protected BagFactory mBagFactory = BagFactory.getInstance();
protected TupleFactory mTupleFactory = TupleFactory.getInstance();
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/MergeJoinIndexer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/MergeJoinIndexer.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/MergeJoinIndexer.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/MergeJoinIndexer.java Tue Oct 27 01:13:19 2009
@@ -21,12 +21,19 @@
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.pig.ExecType;
+import org.apache.pig.LoadCaster;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigException;
import org.apache.pig.SamplableLoader;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
@@ -48,7 +55,7 @@
* (key0, key1,...,fileName, offset)
* These tuples are then sorted before being written out to index file on HDFS.
*/
-
+//XXX : FIXME - make this work with new load-store redesign
public class MergeJoinIndexer implements LoadFunc{
private boolean firstRec = true;
@@ -99,12 +106,6 @@
}
@Override
- public void bindTo(String fileName, BufferedPositionedInputStream is,long offset, long end) throws IOException {
- this.fileName = fileName;
- loader.bindTo(fileName, is, offset, end);
- }
-
- @Override
public Tuple getNext() throws IOException {
if(!firstRec) // We sample only one record per block.
@@ -188,46 +189,48 @@
return wrapperTuple;
}
- public Integer bytesToInteger(byte[] b) throws IOException {
- return loader.bytesToInteger(b);
- }
-
- public Long bytesToLong(byte[] b) throws IOException {
- return loader.bytesToLong(b);
- }
-
- public Float bytesToFloat(byte[] b) throws IOException {
- return loader.bytesToFloat(b);
- }
-
- public Double bytesToDouble(byte[] b) throws IOException {
- return loader.bytesToDouble(b);
- }
-
- public String bytesToCharArray(byte[] b) throws IOException {
- return loader.bytesToCharArray(b);
- }
-
- public Map<String, Object> bytesToMap(byte[] b) throws IOException {
- return loader.bytesToMap(b);
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
}
- public Tuple bytesToTuple(byte[] b) throws IOException {
- return loader.bytesToTuple(b);
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
+ @Override
+ public InputFormat getInputFormat() {
+ // TODO Auto-generated method stub
+ return null;
}
- public DataBag bytesToBag(byte[] b) throws IOException {
- return loader.bytesToBag(b);
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return null;
}
- public void fieldsToRead(Schema schema) {
- loader.fieldsToRead(schema);
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
}
- public Schema determineSchema(
- String fileName,
- ExecType execType,
- DataStorage storage) throws IOException {
- return loader.determineSchema(fileName, execType, storage);
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
}
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/PoissonSampleLoader.java Tue Oct 27 01:13:19 2009
@@ -21,22 +21,26 @@
import java.util.ArrayList;
import java.util.Properties;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.pig.LoadCaster;
import org.apache.pig.PigException;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.Tuple;
-import org.apache.pig.data.TupleFactory;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.impl.PigContext;
-import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.util.Pair;
-import org.apache.pig.impl.builtin.PartitionSkewedKeys;
/**
* Currently skipInterval is similar to the randomsampleloader. However, if we were to use an
* uniform distribution, we could precompute the intervals and read it from a file.
*
*/
+//XXX : FIXME - make this work with new load-store redesign
public class PoissonSampleLoader extends SampleLoader {
// Base number of samples needed
@@ -62,6 +66,7 @@
// By default the data is multiplied by 2 to account for the encoding
private static final int DEFAULT_CONV_FACTOR = 2;
+ private final Log log = LogFactory.getLog(getClass());
public PoissonSampleLoader(String funcSpec, String ns) {
super(funcSpec);
@@ -140,6 +145,42 @@
n = (n > 1) ? n : 1;
setNumSamples(n);
}
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/RandomSampleLoader.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/RandomSampleLoader.java?rev=830041&r1=830040&r2=830041&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/RandomSampleLoader.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/builtin/RandomSampleLoader.java Tue Oct 27 01:13:19 2009
@@ -17,11 +17,21 @@
*/
package org.apache.pig.impl.builtin;
+import java.io.IOException;
+
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.pig.LoadCaster;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+
/**
* A loader that samples the data. This loader can subsume loader that
* can handle starting in the middle of a record. Loaders that can
* handle this should implement the SamplableLoader interface.
*/
+//XXX : FIXME - make this work with new load-store redesign
public class RandomSampleLoader extends SampleLoader {
/**
@@ -45,5 +55,54 @@
// Setting it to 100 as default for order by
super.setNumSamples(100);
}
+
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#doneReading()
+ */
+ @Override
+ public void doneReading() {
+ // TODO Auto-generated method stub
+
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getInputFormat()
+ */
+ @Override
+ public InputFormat getInputFormat() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#getLoadCaster()
+ */
+ @Override
+ public LoadCaster getLoadCaster() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#prepareToRead(org.apache.hadoop.mapreduce.RecordReader, org.apache.hadoop.mapreduce.InputSplit)
+ */
+ @Override
+ public void prepareToRead(RecordReader reader, PigSplit split) {
+ // TODO Auto-generated method stub
+
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.LoadFunc#setLocation(java.lang.String, org.apache.hadoop.mapreduce.Job)
+ */
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}