You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ol...@apache.org on 2008/02/07 00:09:55 UTC
svn commit: r619213 [2/5] - in /incubator/pig/trunk: ./ 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/pig/backend/hadoop/...
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/MapReduceLauncher.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/MapReduceLauncher.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/MapReduceLauncher.java Wed Feb 6 15:09:43 2008
@@ -87,7 +87,7 @@
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){
return WritableComparator.compareBytes(b1, s1, l1, b2, s2, l2);
- }
+ }
}
static Random rand = new Random();
@@ -134,7 +134,7 @@
}
// create jobs.jar locally and pass it to hadoop
- File submitJarFile = File.createTempFile("Job", ".jar");
+ File submitJarFile = File.createTempFile("Job", ".jar");
try {
FileOutputStream fos = new FileOutputStream(submitJarFile);
JarManager.createJar(fos, funcs, pom.pigContext);
@@ -174,7 +174,7 @@
}
else{
// this is not a sort job - can use byte comparison to speed up processing
- conf.setOutputKeyComparatorClass(PigWritableComparator.class);
+ conf.setOutputKeyComparatorClass(PigWritableComparator.class);
}
if (pom.partitionFunction!=null){
conf.setPartitionerClass(SortPartitioner.class);
@@ -198,99 +198,99 @@
// Now, actually submit the job (using the submit name)
//
JobClient jobClient = execEngine.getJobClient();
- RunningJob status = jobClient.submitJob(conf);
- log.debug("submitted job: " + status.getJobID());
+ RunningJob status = jobClient.submitJob(conf);
+ log.debug("submitted job: " + status.getJobID());
- long sleepTime = 1000;
- double lastQueryProgress = -1.0;
- int lastJobsQueued = -1;
- double lastMapProgress = -1.0;
- double lastReduceProgress = -1.0;
- while (true) {
- try {
- Thread.sleep(sleepTime); } catch (Exception e) {}
-
- if (status.isComplete()) {
- success = status.isSuccessful();
- log.debug("Job finished " + (success ? "" : "un") + "successfully");
- if (success) {
- mrJobNumber++;
- }
- double queryProgress = ((double) mrJobNumber) / ((double) numMRJobs);
- if (queryProgress > lastQueryProgress) {
- log.info("Pig progress = " + ((int) (queryProgress * 100)) + "%");
- lastQueryProgress = queryProgress;
- }
- break;
- }
- else // still running
- {
- double mapProgress = status.mapProgress();
- double reduceProgress = status.reduceProgress();
- if (lastMapProgress != mapProgress || lastReduceProgress != reduceProgress) {
- log.debug("Hadoop job progress: Map=" + (int) (mapProgress * 100) + "% Reduce="
- + (int) (reduceProgress * 100) + "%");
- lastMapProgress = mapProgress;
- lastReduceProgress = reduceProgress;
- }
- double numJobsCompleted = mrJobNumber;
- double thisJobProgress = (mapProgress + reduceProgress) / 2.0;
- double queryProgress = (numJobsCompleted + thisJobProgress) / ((double) numMRJobs);
- if (queryProgress > lastQueryProgress) {
- log.info("Pig progress = " + ((int) (queryProgress * 100)) + "%");
- lastQueryProgress = queryProgress;
- }
- }
- }
-
- // bug 1030028: if the input file is empty; hadoop doesn't create the output file!
- Path outputFile = conf.getOutputPath();
- String outputName = outputFile.getName();
- int colon = outputName.indexOf(':');
- if (colon != -1) {
- outputFile = new Path(outputFile.getParent(), outputName.substring(0, colon));
- }
-
- try {
- ElementDescriptor descriptor =
- ((HDataStorage)(pom.pigContext.getDfs())).asElement(outputFile.toString());
+ long sleepTime = 1000;
+ double lastQueryProgress = -1.0;
+ int lastJobsQueued = -1;
+ double lastMapProgress = -1.0;
+ double lastReduceProgress = -1.0;
+ while (true) {
+ try {
+ Thread.sleep(sleepTime); } catch (Exception e) {}
+
+ if (status.isComplete()) {
+ success = status.isSuccessful();
+ log.debug("Job finished " + (success ? "" : "un") + "successfully");
+ if (success) {
+ mrJobNumber++;
+ }
+ double queryProgress = ((double) mrJobNumber) / ((double) numMRJobs);
+ if (queryProgress > lastQueryProgress) {
+ log.info("Pig progress = " + ((int) (queryProgress * 100)) + "%");
+ lastQueryProgress = queryProgress;
+ }
+ break;
+ }
+ else // still running
+ {
+ double mapProgress = status.mapProgress();
+ double reduceProgress = status.reduceProgress();
+ if (lastMapProgress != mapProgress || lastReduceProgress != reduceProgress) {
+ log.debug("Hadoop job progress: Map=" + (int) (mapProgress * 100) + "% Reduce="
+ + (int) (reduceProgress * 100) + "%");
+ lastMapProgress = mapProgress;
+ lastReduceProgress = reduceProgress;
+ }
+ double numJobsCompleted = mrJobNumber;
+ double thisJobProgress = (mapProgress + reduceProgress) / 2.0;
+ double queryProgress = (numJobsCompleted + thisJobProgress) / ((double) numMRJobs);
+ if (queryProgress > lastQueryProgress) {
+ log.info("Pig progress = " + ((int) (queryProgress * 100)) + "%");
+ lastQueryProgress = queryProgress;
+ }
+ }
+ }
+
+ // bug 1030028: if the input file is empty; hadoop doesn't create the output file!
+ Path outputFile = conf.getOutputPath();
+ String outputName = outputFile.getName();
+ int colon = outputName.indexOf(':');
+ if (colon != -1) {
+ outputFile = new Path(outputFile.getParent(), outputName.substring(0, colon));
+ }
+
+ try {
+ ElementDescriptor descriptor =
+ ((HDataStorage)(pom.pigContext.getDfs())).asElement(outputFile.toString());
- if (success && !descriptor.exists()) {
+ if (success && !descriptor.exists()) {
- // create an empty output file
- PigFile f = new PigFile(outputFile.toString(), false);
- f.store(BagFactory.getInstance().newDefaultBag(),
- new PigStorage(),
- pom.pigContext);
- }
- }
- catch (DataStorageException e) {
- IOException ioe = new IOException("Failed to obtain descriptor for " + outputFile.toString());
- ioe.initCause(e);
- throw ioe;
- }
-
- if (!success) {
- // go find the error messages
- getErrorMessages(jobClient.getMapTaskReports(status.getJobID()),
- "map", log);
- getErrorMessages(jobClient.getReduceTaskReports(status.getJobID()),
- "reduce", log);
- }
- else {
- long timeSpent = 0;
+ // create an empty output file
+ PigFile f = new PigFile(outputFile.toString(), false);
+ f.store(BagFactory.getInstance().newDefaultBag(),
+ new PigStorage(),
+ pom.pigContext);
+ }
+ }
+ catch (DataStorageException e) {
+ IOException ioe = new IOException("Failed to obtain descriptor for " + outputFile.toString());
+ ioe.initCause(e);
+ throw ioe;
+ }
+
+ if (!success) {
+ // go find the error messages
+ getErrorMessages(jobClient.getMapTaskReports(status.getJobID()),
+ "map", log);
+ getErrorMessages(jobClient.getReduceTaskReports(status.getJobID()),
+ "reduce", log);
+ }
+ else {
+ long timeSpent = 0;
- // NOTE: this call is crashing due to a bug in Hadoop; the bug is known and the patch has not been applied yet.
- TaskReport[] mapReports = jobClient.getMapTaskReports(status.getJobID());
- TaskReport[] reduceReports = jobClient.getReduceTaskReports(status.getJobID());
- for (TaskReport r : mapReports) {
- timeSpent += (r.getFinishTime() - r.getStartTime());
- }
- for (TaskReport r : reduceReports) {
- timeSpent += (r.getFinishTime() - r.getStartTime());
- }
- totalHadoopTimeSpent += timeSpent;
- }
+ // NOTE: this call is crashing due to a bug in Hadoop; the bug is known and the patch has not been applied yet.
+ TaskReport[] mapReports = jobClient.getMapTaskReports(status.getJobID());
+ TaskReport[] reduceReports = jobClient.getReduceTaskReports(status.getJobID());
+ for (TaskReport r : mapReports) {
+ timeSpent += (r.getFinishTime() - r.getStartTime());
+ }
+ for (TaskReport r : reduceReports) {
+ timeSpent += (r.getFinishTime() - r.getStartTime());
+ }
+ totalHadoopTimeSpent += timeSpent;
+ }
}
catch (Exception e) {
// Do we need different handling for different exceptions
@@ -305,15 +305,15 @@
private void getErrorMessages(TaskReport reports[], String type, Logger log)
{
- for (int i = 0; i < reports.length; i++) {
- String msgs[] = reports[i].getDiagnostics();
- StringBuilder sb = new StringBuilder("Error message from task (" + type + ") " +
- reports[i].getTaskId());
- for (int j = 0; j < msgs.length; j++) {
- sb.append(" " + msgs[j]);
- }
- log.error(sb.toString());
- }
+ for (int i = 0; i < reports.length; i++) {
+ String msgs[] = reports[i].getDiagnostics();
+ StringBuilder sb = new StringBuilder("Error message from task (" + type + ") " +
+ reports[i].getTaskId());
+ for (int j = 0; j < msgs.length; j++) {
+ sb.append(" " + msgs[j]);
+ }
+ log.error(sb.toString());
+ }
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java Wed Feb 6 15:09:43 2008
@@ -123,17 +123,17 @@
public CombineDataOutputCollector(OutputCollector oc) {
super(null);
- this.oc = oc;
+ this.oc = oc;
}
@Override
- public void add(Datum d){
+ public void add(Datum d){
if (d == null) return; // EOF marker from eval pipeline; ignore
try{
- // oc.collect(group, new IndexedTuple(((Tuple)d).getTupleField(0),index));
+ // oc.collect(group, new IndexedTuple(((Tuple)d).getTupleField(0),index));
oc.collect(group, new IndexedTuple(((Tuple)d),index));
}catch (IOException e){
- throw new RuntimeException(e);
+ throw new RuntimeException(e);
}
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java Wed Feb 6 15:09:43 2008
@@ -49,8 +49,8 @@
public class PigInputFormat implements InputFormat<Text, Tuple>, JobConfigurable {
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
-
- ArrayList<FileSpec> inputs = (ArrayList<FileSpec>) ObjectSerializer.deserialize(job.get("pig.inputs"));
+
+ ArrayList<FileSpec> inputs = (ArrayList<FileSpec>) ObjectSerializer.deserialize(job.get("pig.inputs"));
ArrayList<EvalSpec> mapFuncs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.mapFuncs",""));
ArrayList<EvalSpec> groupFuncs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.groupFuncs", ""));
PigContext pigContext = (PigContext)ObjectSerializer.deserialize(job.get("pig.pigContext"));
@@ -75,25 +75,25 @@
ArrayList<InputSplit> splits = new ArrayList<InputSplit>();
for (int i = 0; i < inputs.size(); i++) {
- Path path = new Path(inputs.get(i).getFileName());
+ Path path = new Path(inputs.get(i).getFileName());
String parser = inputs.get(i).getFuncSpec();
FileSystem fs = path.getFileSystem(job);
fs.setWorkingDirectory(new Path("/user", job.getUser()));
ArrayList<Path> paths = new ArrayList<Path>();
- // If you give a non-glob name, globPaths returns a single
- // element with just that name.
- Path[] globPaths = fs.globPaths(path);
- for (int m = 0; m < globPaths.length; m++) paths.add(globPaths[m]);
+ // If you give a non-glob name, globPaths returns a single
+ // element with just that name.
+ Path[] globPaths = fs.globPaths(path);
+ for (int m = 0; m < globPaths.length; m++) paths.add(globPaths[m]);
//paths.add(path);
for (int j = 0; j < paths.size(); j++) {
Path fullPath = new Path(fs.getWorkingDirectory(), paths.get(j));
if (fs.getFileStatus(fullPath).isDir()) {
- FileStatus children[] = fs.listStatus(fullPath);
- for(int k = 0; k < children.length; k++) {
- paths.add(children[k].getPath());
- }
- continue;
+ FileStatus children[] = fs.listStatus(fullPath);
+ for(int k = 0; k < children.length; k++) {
+ paths.add(children[k].getPath());
+ }
+ continue;
}
long bs = fs.getFileStatus(fullPath).getBlockSize();
long size = fs.getFileStatus(fullPath).getLen();
@@ -158,22 +158,22 @@
fs.setWorkingDirectory(new Path("/user/" + job.getUser()));
CompressionCodec codec = compressionFactory.getCodec(split.getPath());
long start = split.getStart();
- fsis = fs.open(split.getPath());
- fsis.seek(start);
+ fsis = fs.open(split.getPath());
+ fsis.seek(start);
if (codec != null) {
is = codec.createInputStream(fsis);
end = Long.MAX_VALUE;
} else{
- end = start + split.getLength();
-
- if (split.file.getName().endsWith(".bz") ||
- split.file.getName().endsWith(".bz2")) {
- is = new CBZip2InputStream(fsis,9);
- }else{
- is = fsis;
- }
+ end = start + split.getLength();
+
+ if (split.file.getName().endsWith(".bz") ||
+ split.file.getName().endsWith(".bz2")) {
+ is = new CBZip2InputStream(fsis,9);
+ }else{
+ is = fsis;
+ }
}
myReader.set(this);
loader.bindTo(split.getPath().toString(), new BufferedPositionedInputStream(is, start), start, end);
@@ -181,7 +181,7 @@
}
public JobConf getJobConf(){
- return job;
+ return job;
}
public boolean next(Text key, Tuple value) throws IOException {
@@ -196,7 +196,7 @@
}
public long getPos() throws IOException {
- return fsis.getPos();
+ return fsis.getPos();
}
public void close() throws IOException {
@@ -215,11 +215,11 @@
return new Tuple();
}
- public float getProgress() throws IOException {
- float progress = getPos() - split.getStart();
- float finish = split.getLength();
- return progress/finish;
- }
+ public float getProgress() throws IOException {
+ float progress = getPos() - split.getStart();
+ float finish = split.getLength();
+ return progress/finish;
+ }
}
public void validateInput(JobConf arg0) throws IOException {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java Wed Feb 6 15:09:43 2008
@@ -165,24 +165,24 @@
* Nothing happens here.
*/
public void close() throws IOException {
- if (evalPipe!=null)
- evalPipe.finishPipe();
+ if (evalPipe!=null)
+ evalPipe.finishPipe();
}
public static PigContext getPigContext() {
- if (pigContext!=null)
- return pigContext;
- try {
- InputStream is = PigMapReduce.class.getResourceAsStream("/pigContext");
- if (is == null) throw new RuntimeException("/pigContext not found!");
- return (PigContext)new ObjectInputStream(is).readObject();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ if (pigContext!=null)
+ return pigContext;
+ try {
+ InputStream is = PigMapReduce.class.getResourceAsStream("/pigContext");
+ if (is == null) throw new RuntimeException("/pigContext not found!");
+ return (PigContext)new ObjectInputStream(is).readObject();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
private String getTaskId(){
- String taskidParts[] = job.get("mapred.task.id").split("_"); // The format is
+ String taskidParts[] = job.get("mapred.task.id").split("_"); // The format is
// tip_job_m_partition_try
return taskidParts[taskidParts.length - 2];
}
@@ -202,15 +202,15 @@
if(splitSpec != null) splitSpec.instantiateFunc(pigContext);
if (splitSpec == null){
- pigWriter = (PigRecordWriter) job.getOutputFormat().getRecordWriter(FileSystem.get(job), job, fileName,
- reporter);
- oc = new OutputCollector() {
- public void collect(WritableComparable key, Writable value) throws IOException {
- pigWriter.write(key, value);
- }
- };
+ pigWriter = (PigRecordWriter) job.getOutputFormat().getRecordWriter(FileSystem.get(job), job, fileName,
+ reporter);
+ oc = new OutputCollector() {
+ public void collect(WritableComparable key, Writable value) throws IOException {
+ pigWriter.write(key, value);
+ }
+ };
}else{
- oc = getSplitCollector(splitSpec);
+ oc = getSplitCollector(splitSpec);
}
evalPipe = evalSpec.setupPipe(new MapDataOutputCollector());
} else {
@@ -222,36 +222,36 @@
}
private OutputCollector getSplitCollector(final SplitSpec splitSpec) throws IOException{
-
- PigOutputFormat outputFormat = (PigOutputFormat)job.getOutputFormat();
-
- for (String name: splitSpec.tempFiles){
- sideFileWriters.add( outputFormat.getRecordWriter(FileSystem.get(job), job, new Path(name), "split-" + getTaskId(), reporter));
- }
- return new OutputCollector(){
- public void collect(WritableComparable key, Writable value) throws IOException {
- Tuple t = (Tuple) value;
- ArrayList<Cond> conditions = splitSpec.conditions;
- for (int i=0; i< conditions.size(); i++){
- Cond cond = conditions.get(i);
- if (cond.eval(t)){
- //System.out.println("Writing " + t + " to condition " + cond);
- sideFileWriters.get(i).write(null, t);
- }
- }
-
- }
- };
+
+ PigOutputFormat outputFormat = (PigOutputFormat)job.getOutputFormat();
+
+ for (String name: splitSpec.tempFiles){
+ sideFileWriters.add( outputFormat.getRecordWriter(FileSystem.get(job), job, new Path(name), "split-" + getTaskId(), reporter));
+ }
+ return new OutputCollector(){
+ public void collect(WritableComparable key, Writable value) throws IOException {
+ Tuple t = (Tuple) value;
+ ArrayList<Cond> conditions = splitSpec.conditions;
+ for (int i=0; i< conditions.size(); i++){
+ Cond cond = conditions.get(i);
+ if (cond.eval(t)){
+ //System.out.println("Writing " + t + " to condition " + cond);
+ sideFileWriters.get(i).write(null, t);
+ }
+ }
+
+ }
+ };
}
private void setupReducePipe() throws IOException {
pigContext = (PigContext)ObjectSerializer.deserialize(job.get("pig.pigContext"));
- EvalSpec evalSpec = (EvalSpec)ObjectSerializer.deserialize(job.get("pig.reduceFunc", ""));
+ EvalSpec evalSpec = (EvalSpec)ObjectSerializer.deserialize(job.get("pig.reduceFunc", ""));
if (evalSpec == null)
- evalSpec = new StarSpec();
+ evalSpec = new StarSpec();
else
- evalSpec.instantiateFunc(pigContext);
+ evalSpec.instantiateFunc(pigContext);
ArrayList<EvalSpec> groupSpecs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.groupFuncs", ""));
@@ -263,7 +263,7 @@
SplitSpec splitSpec = (SplitSpec) ObjectSerializer.deserialize(job.get("pig.splitSpec", ""));
if (splitSpec != null){
- splitSpec.instantiateFunc(pigContext);
+ splitSpec.instantiateFunc(pigContext);
oc = getSplitCollector(splitSpec);
}
@@ -274,63 +274,63 @@
}
public void closeSideFiles(){
- for (PigRecordWriter writer: sideFileWriters){
- try{
- writer.close(reporter);
- }catch(IOException e){
- e.printStackTrace();
- }
- }
+ for (PigRecordWriter writer: sideFileWriters){
+ try{
+ writer.close(reporter);
+ }catch(IOException e){
+ e.printStackTrace();
+ }
+ }
}
class MapDataOutputCollector extends DataCollector {
-
- public MapDataOutputCollector(){
- super(null);
- }
-
+
+ public MapDataOutputCollector(){
+ super(null);
+ }
+
@Override
- public void add(Datum d){
- try{
- if (group == null) {
- oc.collect(null, (Tuple)d);
- } else {
- Datum[] groupAndTuple = LOCogroup.getGroupAndTuple(d);
+ public void add(Datum d){
+ try{
+ if (group == null) {
+ oc.collect(null, (Tuple)d);
+ } else {
+ Datum[] groupAndTuple = LOCogroup.getGroupAndTuple(d);
// wrap group label in a tuple, so it becomes writable.
- oc.collect(new Tuple(groupAndTuple[0]), new IndexedTuple((Tuple)groupAndTuple[1], index));
+ oc.collect(new Tuple(groupAndTuple[0]), new IndexedTuple((Tuple)groupAndTuple[1], index));
}
}catch(IOException e){
- throw new RuntimeException(e);
+ throw new RuntimeException(e);
}
}
@Override
- public void finish(){
- closeSideFiles();
- if (group != null)
- group.finish();
+ public void finish(){
+ closeSideFiles();
+ if (group != null)
+ group.finish();
}
}
class ReduceDataOutputCollector extends DataCollector {
-
- public ReduceDataOutputCollector(){
- super(null);
- }
-
+
+ public ReduceDataOutputCollector(){
+ super(null);
+ }
+
@Override
- public void add(Datum d){
- try{
- //System.out.println("Adding " + d + " to reduce output");
- oc.collect(null, (Tuple)d);
- }catch(IOException e){
- throw new RuntimeException(e);
- }
+ public void add(Datum d){
+ try{
+ //System.out.println("Adding " + d + " to reduce output");
+ oc.collect(null, (Tuple)d);
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
}
@Override
protected void finish(){
- closeSideFiles();
+ closeSideFiles();
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java Wed Feb 6 15:09:43 2008
@@ -43,25 +43,25 @@
public RecordWriter getRecordWriter(FileSystem fs, JobConf job, String name, Progressable progress)
throws IOException {
- Path outputDir = job.getOutputPath();
+ Path outputDir = job.getOutputPath();
return getRecordWriter(fs, job, outputDir, name, progress);
}
public PigRecordWriter getRecordWriter(FileSystem fs, JobConf job, Path outputDir, String name, Progressable progress)
throws IOException {
- StoreFunc store;
- String storeFunc = job.get("pig.storeFunc", "");
- if (storeFunc.length() == 0) {
- store = new PigStorage();
- } else {
- try {
- store = (StoreFunc) PigContext.instantiateFuncFromSpec(storeFunc);
- } catch (Exception e) {
- RuntimeException re = new RuntimeException(e.getClass().getName() + ": " + e.getMessage());
- re.setStackTrace(e.getStackTrace());
- throw re;
- }
- }
+ StoreFunc store;
+ String storeFunc = job.get("pig.storeFunc", "");
+ if (storeFunc.length() == 0) {
+ store = new PigStorage();
+ } else {
+ try {
+ store = (StoreFunc) PigContext.instantiateFuncFromSpec(storeFunc);
+ } catch (Exception e) {
+ RuntimeException re = new RuntimeException(e.getClass().getName() + ": " + e.getMessage());
+ re.setStackTrace(e.getStackTrace());
+ throw re;
+ }
+ }
String parentName = outputDir.getParent().getName();
int suffixStart = parentName.lastIndexOf('.');
if (suffixStart != -1) {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java Wed Feb 6 15:09:43 2008
@@ -52,15 +52,15 @@
public PigSplit() {}
public PigSplit(PigContext pigContext, FileSystem fs, Path path, String parser, EvalSpec groupbySpec, EvalSpec evalSpec, int index, long start, long length) {
- this.fs = fs;
- this.file = path;
- this.start = start;
- this.length = length;
- this.groupbySpec = groupbySpec;
- this.evalSpec = evalSpec;
- this.index = index;
- this.parser = parser;
- this.pigContext = pigContext;
+ this.fs = fs;
+ this.file = path;
+ this.start = start;
+ this.length = length;
+ this.groupbySpec = groupbySpec;
+ this.evalSpec = evalSpec;
+ this.index = index;
+ this.parser = parser;
+ this.pigContext = pigContext;
}
public String getParser() {
@@ -68,11 +68,11 @@
}
public long getStart() {
- return start;
+ return start;
}
public long getLength() {
- return length;
+ return length;
}
public EvalSpec getGroupbySpec() {
return groupbySpec;
@@ -106,65 +106,65 @@
public String[] getLocations() throws IOException {
- String hints[][] = fs.getFileCacheHints(file, start, length);
- int total = 0;
- for(int i = 0; i < hints.length; i++) {
- total += hints[i].length;
- }
- String locations[] = new String[total];
- int count = 0;
- for(int i = 0; i < hints.length; i++) {
- for(int j = 0; j < hints[i].length; j++) {
- locations[count++] = hints[i][j];
- }
- }
- return locations;
+ String hints[][] = fs.getFileCacheHints(file, start, length);
+ int total = 0;
+ for(int i = 0; i < hints.length; i++) {
+ total += hints[i].length;
+ }
+ String locations[] = new String[total];
+ int count = 0;
+ for(int i = 0; i < hints.length; i++) {
+ for(int j = 0; j < hints[i].length; j++) {
+ locations[count++] = hints[i][j];
+ }
+ }
+ return locations;
}
public void readFields(DataInput is) throws IOException {
- file = new Path(is.readUTF());
- start = is.readLong();
- length = is.readLong();
- pigContext = (PigContext)readObject(is);
- groupbySpec = (EvalSpec)readObject(is);
- if(groupbySpec != null ) groupbySpec.instantiateFunc(pigContext);
- evalSpec = (EvalSpec) readObject(is);
- if(evalSpec != null) evalSpec.instantiateFunc(pigContext);
- index = is.readInt();
- parser = is.readUTF();
+ file = new Path(is.readUTF());
+ start = is.readLong();
+ length = is.readLong();
+ pigContext = (PigContext)readObject(is);
+ groupbySpec = (EvalSpec)readObject(is);
+ if(groupbySpec != null ) groupbySpec.instantiateFunc(pigContext);
+ evalSpec = (EvalSpec) readObject(is);
+ if(evalSpec != null) evalSpec.instantiateFunc(pigContext);
+ index = is.readInt();
+ parser = is.readUTF();
}
public void write(DataOutput os) throws IOException {
- os.writeUTF(file.toString());
- os.writeLong(start);
- os.writeLong(length);
- writeObject(pigContext, os);
- writeObject(groupbySpec, os);
- writeObject(evalSpec, os);
- os.writeInt(index);
- os.writeUTF(parser);
+ os.writeUTF(file.toString());
+ os.writeLong(start);
+ os.writeLong(length);
+ writeObject(pigContext, os);
+ writeObject(groupbySpec, os);
+ writeObject(evalSpec, os);
+ os.writeInt(index);
+ os.writeUTF(parser);
}
private void writeObject(Serializable obj, DataOutput os) throws IOException{
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- ObjectOutputStream oos = new ObjectOutputStream(baos);
- oos.writeObject(obj);
- byte[] bytes = baos.toByteArray();
- os.writeInt(bytes.length);
- os.write(bytes);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ oos.writeObject(obj);
+ byte[] bytes = baos.toByteArray();
+ os.writeInt(bytes.length);
+ os.write(bytes);
}
private Object readObject(DataInput is) throws IOException{
- byte[] bytes = new byte[is.readInt()];
- is.readFully(bytes);
- ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
- try{
- return ois.readObject();
- }catch (ClassNotFoundException cnfe){
- IOException newE = new IOException(cnfe.getMessage());
+ byte[] bytes = new byte[is.readInt()];
+ is.readFully(bytes);
+ ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
+ try{
+ return ois.readObject();
+ }catch (ClassNotFoundException cnfe){
+ IOException newE = new IOException(cnfe.getMessage());
newE.initCause(cnfe);
throw newE;
- }
+ }
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java Wed Feb 6 15:09:43 2008
@@ -34,47 +34,47 @@
public class SortPartitioner implements Partitioner {
- Tuple[] quantiles;
- WritableComparator comparator;
-
- public int getPartition(WritableComparable key, Writable value,
- int numPartitions) {
- try{
- Tuple keyTuple = (Tuple)key;
- int index = Arrays.binarySearch(quantiles, keyTuple.getTupleField(0), comparator);
- if (index < 0)
- index = -index-1;
- return Math.min(index, numPartitions - 1);
- }catch(IOException e){
- throw new RuntimeException(e);
- }
- }
+ Tuple[] quantiles;
+ WritableComparator comparator;
+
+ public int getPartition(WritableComparable key, Writable value,
+ int numPartitions) {
+ try{
+ Tuple keyTuple = (Tuple)key;
+ int index = Arrays.binarySearch(quantiles, keyTuple.getTupleField(0), comparator);
+ if (index < 0)
+ index = -index-1;
+ return Math.min(index, numPartitions - 1);
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
+ }
- public void configure(JobConf job) {
- String quantilesFile = job.get("pig.quantilesFile", "");
- if (quantilesFile.length() == 0)
- throw new RuntimeException("Sort paritioner used but no quantiles found");
-
- try{
- InputStream is = FileLocalizer.openDFSFile(quantilesFile,job);
- BinStorage loader = new BinStorage();
- loader.bindTo(quantilesFile, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
-
- Tuple t;
- ArrayList<Tuple> quantiles = new ArrayList<Tuple>();
-
- while(true){
- t = loader.getNext();
- if (t==null)
- break;
- quantiles.add(t);
- }
- this.quantiles = quantiles.toArray(new Tuple[0]);
- }catch (IOException e){
- throw new RuntimeException(e);
- }
+ public void configure(JobConf job) {
+ String quantilesFile = job.get("pig.quantilesFile", "");
+ if (quantilesFile.length() == 0)
+ throw new RuntimeException("Sort paritioner used but no quantiles found");
+
+ try{
+ InputStream is = FileLocalizer.openDFSFile(quantilesFile,job);
+ BinStorage loader = new BinStorage();
+ loader.bindTo(quantilesFile, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
+
+ Tuple t;
+ ArrayList<Tuple> quantiles = new ArrayList<Tuple>();
+
+ while(true){
+ t = loader.getNext();
+ if (t==null)
+ break;
+ quantiles.add(t);
+ }
+ this.quantiles = quantiles.toArray(new Tuple[0]);
+ }catch (IOException e){
+ throw new RuntimeException(e);
+ }
- comparator = job.getOutputKeyComparator();
- }
+ comparator = job.getOutputKeyComparator();
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java Wed Feb 6 15:09:43 2008
@@ -37,7 +37,7 @@
}
public Map<String, Object> getStatistics() throws IOException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public LocalPath asElement(String name)
@@ -113,7 +113,7 @@
}
public void setActiveContainer(ContainerDescriptor container) {
- this.workingDir = new File(container.toString());
+ this.workingDir = new File(container.toString());
}
public ContainerDescriptor getActiveContainer() {
@@ -121,7 +121,7 @@
}
public LocalPath[] asCollection(String pattern) throws DataStorageException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public File getWorkingDir() {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java Wed Feb 6 15:09:43 2008
@@ -58,16 +58,16 @@
Properties dstConfiguration,
boolean removeSrc)
throws IOException {
- try {
- if (! dstName.getDataStorage().isContainer(dstName.toString())) {
- dstName = dstName.getDataStorage().asContainer(dstName.toString());
- }
- }
- catch (DataStorageException e) {
- IOException ioe = new IOException("Failed to get container for " + dstName.toString());
- ioe.initCause(e);
- throw ioe;
- }
+ try {
+ if (! dstName.getDataStorage().isContainer(dstName.toString())) {
+ dstName = dstName.getDataStorage().asContainer(dstName.toString());
+ }
+ }
+ catch (DataStorageException e) {
+ IOException ioe = new IOException("Failed to get container for " + dstName.toString());
+ ioe.initCause(e);
+ throw ioe;
+ }
copy((ContainerDescriptor) dstName,
dstConfiguration,
@@ -133,7 +133,7 @@
}
public InputStream open(Properties configuration) throws IOException {
- return open();
+ return open();
}
public InputStream open() throws IOException {
@@ -141,7 +141,7 @@
}
public SeekableInputStream sopen(Properties configuration) throws IOException {
- return sopen();
+ return sopen();
}
public SeekableInputStream sopen() throws IOException {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java Wed Feb 6 15:09:43 2008
@@ -100,7 +100,7 @@
}
public InputStream open (Properties configuration) throws IOException {
- return open();
+ return open();
}
public InputStream open () throws IOException {
@@ -108,7 +108,7 @@
}
public SeekableInputStream sopen(Properties configuration) throws IOException {
- return sopen();
+ return sopen();
}
public SeekableInputStream sopen() throws IOException {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java Wed Feb 6 15:09:43 2008
@@ -18,17 +18,17 @@
protected File path;
protected File getCurPath() {
- File path;
-
- if (this.path.isAbsolute()) {
- path = this.path;
- }
- else {
- path = new File(fs.getActiveContainer().toString(),
- this.path.getPath());
- }
-
- return path;
+ File path;
+
+ if (this.path.isAbsolute()) {
+ path = this.path;
+ }
+ else {
+ path = new File(fs.getActiveContainer().toString(),
+ this.path.getPath());
+ }
+
+ return path;
}
public LocalPath(LocalDataStorage fs, String path) {
@@ -49,7 +49,7 @@
public LocalPath(LocalDataStorage fs, File parent, File child) {
this.fs = fs;
this.path = new File(parent.getPath(),
- child.getPath());
+ child.getPath());
}
public LocalPath(LocalDataStorage fs, File parent, String child) {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java Wed Feb 6 15:09:43 2008
@@ -76,7 +76,7 @@
}
public Map<String, Object> getStatistics() throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
@@ -149,11 +149,11 @@
}
public Collection<String> activeScopes() throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public void reclaimScope(String scope) throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
private OperatorKey doCompile(OperatorKey logicalKey,
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java Wed Feb 6 15:09:43 2008
@@ -38,7 +38,7 @@
}
public Map<String, Object> getStatistics() {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public void completionNotification(Object cookie) {
@@ -50,14 +50,14 @@
}
public void getLogs(OutputStream log) throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public void getSTDOut(OutputStream out) throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public void getSTDError(OutputStream error) throws ExecException {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java Wed Feb 6 15:09:43 2008
@@ -24,7 +24,7 @@
}
public Properties getConfiguration() {
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException();
}
public void updateConfiguration(Properties configuration)
@@ -39,7 +39,7 @@
}
public Map<OperatorKey, ExecPhysicalOperator> getOpTable() {
- return opTable;
+ return opTable;
}
public OperatorKey getRoot() {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POCogroup.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POCogroup.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POCogroup.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POCogroup.java Wed Feb 6 15:09:43 2008
@@ -41,10 +41,10 @@
// n-ary, blocking operator. Output has schema: < group_label, { <1>, <2>, ... }, { <a>, <b>, ... } >
class POCogroup extends PhysicalOperator {
/**
- *
- */
- private static final long serialVersionUID = 1L;
- List<Datum[]>[] sortedInputs;
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ List<Datum[]>[] sortedInputs;
List<EvalSpec> specs;
public POCogroup(String scope,
@@ -52,14 +52,14 @@
Map<OperatorKey, ExecPhysicalOperator> opTable,
List<EvalSpec> specs,
int outputType) {
- super(scope, id, opTable, outputType);
+ super(scope, id, opTable, outputType);
this.inputs = new OperatorKey[specs.size()];
this.specs = specs;
}
// drain all inputs, and sort each by group (remember, this is a blocking operator!)
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings("unchecked")
public boolean open() throws IOException {
if (!super.open())
return false;
@@ -67,15 +67,15 @@
sortedInputs = new ArrayList[inputs.length];
for (int i = 0; i < inputs.length; i++) {
-
- final int finalI = i;
+
+ final int finalI = i;
sortedInputs[i] = new ArrayList<Datum[]>();
DataCollector outputFromSpec = new DataCollector(null){
- @Override
- public void add(Datum d) {
- sortedInputs[finalI].add(LOCogroup.getGroupAndTuple(d));
- }
+ @Override
+ public void add(Datum d) {
+ sortedInputs[finalI].add(LOCogroup.getGroupAndTuple(d));
+ }
};
DataCollector inputToSpec = specs.get(i).setupPipe(outputFromSpec);
@@ -97,7 +97,7 @@
}
@Override
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
while (true) { // loop until we find a tuple we're allowed to output (or we hit the end)
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POEval.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POEval.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POEval.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POEval.java Wed Feb 6 15:09:43 2008
@@ -32,10 +32,10 @@
// unary, non-blocking operator.
class POEval extends PhysicalOperator {
/**
- *
- */
- private static final long serialVersionUID = 1L;
- EvalSpec spec;
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ EvalSpec spec;
private DataBuffer buf = null;
private DataCollector evalPipeline = null;
private boolean inputDrained;
@@ -47,8 +47,8 @@
OperatorKey input,
EvalSpec specIn,
int outputType) {
- super(scope, id, opTable, outputType);
- inputs = new OperatorKey[1];
+ super(scope, id, opTable, outputType);
+ inputs = new OperatorKey[1];
inputs[0] = input;
spec = specIn;
@@ -59,7 +59,7 @@
Map<OperatorKey, ExecPhysicalOperator> opTable,
EvalSpec specIn,
int outputType) {
- super(scope, id, opTable, outputType);
+ super(scope, id, opTable, outputType);
inputs = new OperatorKey[1];
inputs[0] = null;
@@ -67,13 +67,13 @@
}
@Override
- public boolean open() throws IOException {
+ public boolean open() throws IOException {
if (!super.open()) return false;
if (buf==null)
- buf = new DataBuffer();
+ buf = new DataBuffer();
if (evalPipeline == null)
- evalPipeline = spec.setupPipe(buf);
+ evalPipeline = spec.setupPipe(buf);
inputDrained = false;
@@ -81,32 +81,32 @@
}
@Override
- public Tuple getNext() throws IOException {
- while (true) {
- if (buf.isEmpty()){
+ public Tuple getNext() throws IOException {
+ while (true) {
+ if (buf.isEmpty()){
// no pending outputs, so look to input to provide more food
if (inputDrained){
return null;
}
-
+
Tuple nextTuple = ((PhysicalOperator)opTable.get(inputs[0])).getNext();
if (nextTuple == null){
inputDrained = true;
evalPipeline.finishPipe();
}else{
- evalPipeline.add(nextTuple);
- lastAdded = nextTuple; // for lineage bookkeeping
+ evalPipeline.add(nextTuple);
+ lastAdded = nextTuple; // for lineage bookkeeping
+ }
+ }else{
+ Tuple output = (Tuple)buf.removeFirst();
+ if (lineageTracer != null) {
+ lineageTracer.insert(output);
+ if (lastAdded != null) lineageTracer.union(lastAdded, output); // update lineage (assumes one-to-many relationship between tuples added to pipeline and output!!)
}
- }else{
- Tuple output = (Tuple)buf.removeFirst();
- if (lineageTracer != null) {
- lineageTracer.insert(output);
- if (lastAdded != null) lineageTracer.union(lastAdded, output); // update lineage (assumes one-to-many relationship between tuples added to pipeline and output!!)
- }
- return output;
- }
+ return output;
+ }
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POLoad.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POLoad.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POLoad.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POLoad.java Wed Feb 6 15:09:43 2008
@@ -34,10 +34,10 @@
public class POLoad extends PhysicalOperator {
/**
- *
- */
- private static final long serialVersionUID = 1L;
- String filename;
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ String filename;
LoadFunc lf;
boolean bound = false;
PigContext pigContext;
@@ -48,25 +48,25 @@
PigContext pigContext,
FileSpec fileSpec,
int outputType) {
- super(scope, id, opTable, outputType);
+ super(scope, id, opTable, outputType);
inputs = new OperatorKey[0];
filename = fileSpec.getFileName();
try{
- lf = (LoadFunc) PigContext.instantiateFuncFromSpec(fileSpec.getFuncSpec());
+ lf = (LoadFunc) PigContext.instantiateFuncFromSpec(fileSpec.getFuncSpec());
}catch(Exception e){
- throw new RuntimeException(e);
+ throw new RuntimeException(e);
}
this.pigContext = pigContext;
}
@Override
- public boolean open() throws IOException {
- if (!bound){
- lf.bindTo(filename, new BufferedPositionedInputStream(FileLocalizer.open(filename, pigContext)), 0, Long.MAX_VALUE);
- bound = true;
- }
+ public boolean open() throws IOException {
+ if (!bound){
+ lf.bindTo(filename, new BufferedPositionedInputStream(FileLocalizer.open(filename, pigContext)), 0, Long.MAX_VALUE);
+ bound = true;
+ }
return true;
}
@@ -77,7 +77,7 @@
}
@Override
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
return lf.getNext();
}
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSort.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSort.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSort.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSort.java Wed Feb 6 15:09:43 2008
@@ -33,42 +33,42 @@
public class POSort extends PhysicalOperator {
- static final long serialVersionUID = 1L;
- EvalSpec sortSpec;
- transient Iterator<Tuple> iter;
-
-
- public POSort(String scope,
- long id,
- Map<OperatorKey, ExecPhysicalOperator> opTable,
- EvalSpec sortSpec,
- int outputType) {
- super(scope, id, opTable, outputType);
- this.sortSpec = sortSpec;
- this.inputs = new OperatorKey[1];
- }
+ static final long serialVersionUID = 1L;
+ EvalSpec sortSpec;
+ transient Iterator<Tuple> iter;
+
+
+ public POSort(String scope,
+ long id,
+ Map<OperatorKey, ExecPhysicalOperator> opTable,
+ EvalSpec sortSpec,
+ int outputType) {
+ super(scope, id, opTable, outputType);
+ this.sortSpec = sortSpec;
+ this.inputs = new OperatorKey[1];
+ }
- @Override
- public boolean open() throws IOException {
- if (!super.open())
- return false;
- DataBag bag = BagFactory.getInstance().newSortedBag(sortSpec);
-
- Tuple t;
- while((t = ((PhysicalOperator)opTable.get(inputs[0])).getNext())!=null){
- bag.add(t);
- }
- iter = bag.iterator();
- return true;
- }
-
- @Override
- public Tuple getNext() throws IOException {
- if (iter.hasNext())
- return iter.next();
- else
- return null;
- }
+ @Override
+ public boolean open() throws IOException {
+ if (!super.open())
+ return false;
+ DataBag bag = BagFactory.getInstance().newSortedBag(sortSpec);
+
+ Tuple t;
+ while((t = ((PhysicalOperator)opTable.get(inputs[0])).getNext())!=null){
+ bag.add(t);
+ }
+ iter = bag.iterator();
+ return true;
+ }
+
+ @Override
+ public Tuple getNext() throws IOException {
+ if (iter.hasNext())
+ return iter.next();
+ else
+ return null;
+ }
@Override
public void visit(POVisitor v, String prefix) {
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSplit.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSplit.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSplit.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POSplit.java Wed Feb 6 15:09:43 2008
@@ -33,22 +33,22 @@
protected ArrayList<Cond> conditions;
protected int readFromCond;
- public POSplit(String scope,
- long id,
- Map<OperatorKey, ExecPhysicalOperator> opTable,
- OperatorKey input,
- ArrayList<Cond> conditions,
- int readFromCond,
- int outputType){
- super(scope, id, opTable, outputType);
-
- this.inputs = new OperatorKey[1];
- this.inputs[0] = input;
-
- this.conditions = conditions;
- this.readFromCond = readFromCond;
- }
-
+ public POSplit(String scope,
+ long id,
+ Map<OperatorKey, ExecPhysicalOperator> opTable,
+ OperatorKey input,
+ ArrayList<Cond> conditions,
+ int readFromCond,
+ int outputType){
+ super(scope, id, opTable, outputType);
+
+ this.inputs = new OperatorKey[1];
+ this.inputs[0] = input;
+
+ this.conditions = conditions;
+ this.readFromCond = readFromCond;
+ }
+
@Override
public boolean open() throws IOException{
if (!super.open()){
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POStore.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POStore.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POStore.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POStore.java Wed Feb 6 15:09:43 2008
@@ -36,10 +36,10 @@
public class POStore extends PhysicalOperator {
/**
- *
- */
- private static final long serialVersionUID = 1L;
- private PigFile f;
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ private PigFile f;
private String funcSpec;
boolean append = false;
PigContext pigContext;
@@ -56,8 +56,8 @@
FileSpec outputFileSpec,
boolean append,
PigContext pigContext) {
- super(scope, id, opTable, LogicalOperator.FIXED);
- funcSpec = outputFileSpec.getFuncSpec();
+ super(scope, id, opTable, LogicalOperator.FIXED);
+ funcSpec = outputFileSpec.getFuncSpec();
inputs = new OperatorKey[1];
inputs[0] = input;
System.out.println("Creating " + outputFileSpec.getFileName());
@@ -77,7 +77,7 @@
FileSpec outputFileSpec,
boolean append,
PigContext pigContext) {
- super(scope, id, opTable, LogicalOperator.FIXED);
+ super(scope, id, opTable, LogicalOperator.FIXED);
inputs = new OperatorKey[1];
inputs[0] = null;
funcSpec = outputFileSpec.getFuncSpec();
@@ -90,7 +90,7 @@
}
@Override
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
// get all tuples from input, and store them.
DataBag b = BagFactory.getInstance().newDefaultBag();
Tuple t;
@@ -98,30 +98,30 @@
b.add(t);
}
try {
- StoreFunc func = (StoreFunc) PigContext.instantiateFuncFromSpec(funcSpec);
- f.store(b, func, pigContext);
-
- // a result has materialized, track it!
- LocalResult materializedResult = new LocalResult(this.outFileSpec);
+ StoreFunc func = (StoreFunc) PigContext.instantiateFuncFromSpec(funcSpec);
+ f.store(b, func, pigContext);
+
+ // a result has materialized, track it!
+ LocalResult materializedResult = new LocalResult(this.outFileSpec);
- materializedResults.put(logicalKey, materializedResult);
+ materializedResults.put(logicalKey, materializedResult);
} catch(IOException e) {
- throw e;
+ throw e;
} catch(Exception e) {
- IOException ne = new IOException(e.getClass().getName() + ": " + e.getMessage());
- ne.setStackTrace(e.getStackTrace());
- throw ne;
+ IOException ne = new IOException(e.getClass().getName() + ": " + e.getMessage());
+ ne.setStackTrace(e.getStackTrace());
+ throw ne;
}
return null;
}
@Override
- public int getOutputType(){
- System.err.println("No one should be asking my output type");
- RuntimeException runtimeException = new RuntimeException();
- runtimeException.printStackTrace();
- throw runtimeException;
+ public int getOutputType(){
+ System.err.println("No one should be asking my output type");
+ RuntimeException runtimeException = new RuntimeException();
+ runtimeException.printStackTrace();
+ throw runtimeException;
}
@Override
Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POUnion.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POUnion.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POUnion.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/POUnion.java Wed Feb 6 15:09:43 2008
@@ -30,43 +30,43 @@
class POUnion extends PhysicalOperator {
/**
- *
- */
- private static final long serialVersionUID = 1L;
- int currentInput;
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ int currentInput;
public POUnion(String scope,
long id,
Map<OperatorKey, ExecPhysicalOperator> opTable,
OperatorKey[] inputsIn,
int outputType) {
- super(scope, id, opTable, outputType);
+ super(scope, id, opTable, outputType);
inputs = inputsIn;
currentInput = 0;
}
public POUnion(String scope,
- long id,
- Map<OperatorKey, ExecPhysicalOperator> opTable,
- int numInputs,
- int outputType) {
- super(scope, id, opTable, outputType);
+ long id,
+ Map<OperatorKey, ExecPhysicalOperator> opTable,
+ int numInputs,
+ int outputType) {
+ super(scope, id, opTable, outputType);
inputs = new OperatorKey[numInputs];
for (int i = 0; i < inputs.length; i++)
inputs[i] = null;
}
@Override
- public boolean open() throws IOException{
- if (!super.open()){
- return false;
- }
- currentInput = 0;
- return true;
+ public boolean open() throws IOException{
+ if (!super.open()){
+ return false;
+ }
+ currentInput = 0;
+ return true;
}
@Override
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
while (currentInput < inputs.length) {
Tuple t = ((PhysicalOperator)opTable.get(inputs[currentInput])).getNext();
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/AVG.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/AVG.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/AVG.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/AVG.java Wed Feb 6 15:09:43 2008
@@ -34,9 +34,9 @@
* implemenation, so if possible the execution will be split into a local and global application
*/
public class AVG extends EvalFunc<DataAtom> implements Algebraic {
-
+
@Override
- public void exec(Tuple input, DataAtom output) throws IOException {
+ public void exec(Tuple input, DataAtom output) throws IOException {
double sum = sum(input);
double count = count(input);
@@ -61,27 +61,27 @@
static public class Initial extends EvalFunc<Tuple> {
@Override
- public void exec(Tuple input, Tuple output) throws IOException {
- try {
+ public void exec(Tuple input, Tuple output) throws IOException {
+ try {
output.appendField(new DataAtom(sum(input)));
output.appendField(new DataAtom(count(input)));
// output.appendField(new DataAtom("processed by initial"));
- } catch(RuntimeException t) {
- throw new RuntimeException(t.getMessage() + ": " + input);
- }
+ } catch(RuntimeException t) {
+ throw new RuntimeException(t.getMessage() + ": " + input);
+ }
}
}
static public class Intermed extends EvalFunc<Tuple> {
@Override
- public void exec(Tuple input, Tuple output) throws IOException {
+ public void exec(Tuple input, Tuple output) throws IOException {
combine(input.getBagField(0), output);
}
}
static public class Final extends EvalFunc<DataAtom> {
@Override
- public void exec(Tuple input, DataAtom output) throws IOException {
+ public void exec(Tuple input, DataAtom output) throws IOException {
Tuple combined = new Tuple();
if(input.getField(0) instanceof DataBag) {
combine(input.getBagField(0), combined);
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/BinStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/BinStorage.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/BinStorage.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/BinStorage.java Wed Feb 6 15:09:43 2008
@@ -42,12 +42,12 @@
public BinStorage() {
}
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
byte b = 0;
// skip to next record
while (true) {
- if (in == null || in.getPosition() >=end) {
+ if (in == null || in.getPosition() >=end) {
return null;
}
b = (byte) in.read();
@@ -72,7 +72,7 @@
return t;
}
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
+ public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
this.in = in;
inData = new DataInputStream(in);
this.end = end;
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/COUNT.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/COUNT.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/COUNT.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/COUNT.java Wed Feb 6 15:09:43 2008
@@ -37,7 +37,7 @@
public class COUNT extends EvalFunc<DataAtom> implements Algebraic{
@Override
- public void exec(Tuple input, DataAtom output) throws IOException {
+ public void exec(Tuple input, DataAtom output) throws IOException {
output.setValue(count(input));
}
@@ -55,21 +55,21 @@
static public class Initial extends EvalFunc<Tuple> {
@Override
- public void exec(Tuple input, Tuple output) throws IOException {
+ public void exec(Tuple input, Tuple output) throws IOException {
output.appendField(new DataAtom(count(input)));
}
}
static public class Intermed extends EvalFunc<Tuple> {
@Override
- public void exec(Tuple input, Tuple output) throws IOException {
+ public void exec(Tuple input, Tuple output) throws IOException {
output.appendField(new DataAtom(sum(input)));
}
}
static public class Final extends EvalFunc<DataAtom> {
@Override
- public void exec(Tuple input, DataAtom output) throws IOException {
+ public void exec(Tuple input, DataAtom output) throws IOException {
output.setValue(sum(input));
}
}
@@ -77,11 +77,11 @@
static protected long count(Tuple input) throws IOException {
Datum values = input.getField(0);
if (values instanceof DataBag)
- return ((DataBag)values).size();
+ return ((DataBag)values).size();
else if (values instanceof DataMap)
- return ((DataMap)values).cardinality();
+ return ((DataMap)values).cardinality();
else
- throw new IOException("Cannot count a " + values.getClass().getSimpleName());
+ throw new IOException("Cannot count a " + values.getClass().getSimpleName());
}
static protected long sum(Tuple input) throws IOException {
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/IsEmpty.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/IsEmpty.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/IsEmpty.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/IsEmpty.java Wed Feb 6 15:09:43 2008
@@ -30,13 +30,13 @@
@Override
public boolean exec(Tuple input) throws IOException {
- Datum values = input.getField(0);
+ Datum values = input.getField(0);
if (values instanceof DataBag)
- return ((DataBag)values).size() == 0;
+ return ((DataBag)values).size() == 0;
else if (values instanceof DataMap)
- return ((DataMap)values).cardinality() == 0;
+ return ((DataMap)values).cardinality() == 0;
else
- throw new IOException("Cannot test a " + values.getClass().getSimpleName() + " for emptiness.");
+ throw new IOException("Cannot test a " + values.getClass().getSimpleName() + " for emptiness.");
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/MAX.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/MAX.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/MAX.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/MAX.java Wed Feb 6 15:09:43 2008
@@ -34,58 +34,58 @@
*/
public class MAX extends EvalFunc<DataAtom> implements Algebraic {
- @Override
- public void exec(Tuple input, DataAtom output) throws IOException {
- output.setValue(max(input));
- }
-
- public String getInitial() {
- return Initial.class.getName();
- }
-
- public String getIntermed() {
- return Initial.class.getName();
- }
-
- public String getFinal() {
- return Final.class.getName();
- }
-
- static public class Initial extends EvalFunc<Tuple> {
- @Override
- public void exec(Tuple input, Tuple output) throws IOException {
- output.appendField(new DataAtom(max(input)));
- }
- }
- static public class Final extends EvalFunc<DataAtom> {
- @Override
- public void exec(Tuple input, DataAtom output) throws IOException {
- output.setValue(max(input));
- }
- }
-
- static protected double max(Tuple input) throws IOException {
- DataBag values = input.getBagField(0);
-
- double curMax = Double.NEGATIVE_INFINITY;
- for (Iterator it = values.iterator(); it.hasNext();) {
- Tuple t = (Tuple) it.next();
- try {
- curMax = java.lang.Math.max(curMax, t.getAtomField(0).numval());
- }catch(RuntimeException exp) {
- IOException newE = new IOException("Error processing: " + t.toString() + exp.getMessage());
+ @Override
+ public void exec(Tuple input, DataAtom output) throws IOException {
+ output.setValue(max(input));
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ @Override
+ public void exec(Tuple input, Tuple output) throws IOException {
+ output.appendField(new DataAtom(max(input)));
+ }
+ }
+ static public class Final extends EvalFunc<DataAtom> {
+ @Override
+ public void exec(Tuple input, DataAtom output) throws IOException {
+ output.setValue(max(input));
+ }
+ }
+
+ static protected double max(Tuple input) throws IOException {
+ DataBag values = input.getBagField(0);
+
+ double curMax = Double.NEGATIVE_INFINITY;
+ for (Iterator it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple) it.next();
+ try {
+ curMax = java.lang.Math.max(curMax, t.getAtomField(0).numval());
+ }catch(RuntimeException exp) {
+ IOException newE = new IOException("Error processing: " + t.toString() + exp.getMessage());
newE.initCause(exp);
throw newE;
- }
- }
+ }
+ }
- return curMax;
- }
- @Override
- public Schema outputSchema(Schema input) {
- return new AtomSchema("max" + count++);
- }
+ return curMax;
+ }
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new AtomSchema("max" + count++);
+ }
- private static int count = 1;
+ private static int count = 1;
}
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/MIN.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/MIN.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/MIN.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/MIN.java Wed Feb 6 15:09:43 2008
@@ -34,57 +34,57 @@
*/
public class MIN extends EvalFunc<DataAtom> implements Algebraic {
- @Override
- public void exec(Tuple input, DataAtom output) throws IOException {
- output.setValue(min(input));
- }
-
- public String getInitial() {
- return Initial.class.getName();
- }
-
- public String getIntermed() {
- return Initial.class.getName();
- }
-
- public String getFinal() {
- return Final.class.getName();
- }
-
- static public class Initial extends EvalFunc<Tuple> {
- @Override
- public void exec(Tuple input, Tuple output) throws IOException {
- output.appendField(new DataAtom(min(input)));
- }
- }
- static public class Final extends EvalFunc<DataAtom> {
- @Override
- public void exec(Tuple input, DataAtom output) throws IOException {
- output.setValue(min(input));
- }
- }
-
- static protected double min(Tuple input) throws IOException {
- DataBag values = input.getBagField(0);
-
- double curMin = Double.POSITIVE_INFINITY;
- for (Iterator it = values.iterator(); it.hasNext();) {
- Tuple t = (Tuple) it.next();
- try {
- curMin = java.lang.Math.min(curMin, t.getAtomField(0).numval());
- }catch(RuntimeException exp) {
- IOException newE = new IOException("Error processing: " + t.toString() + exp.getMessage());
+ @Override
+ public void exec(Tuple input, DataAtom output) throws IOException {
+ output.setValue(min(input));
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ @Override
+ public void exec(Tuple input, Tuple output) throws IOException {
+ output.appendField(new DataAtom(min(input)));
+ }
+ }
+ static public class Final extends EvalFunc<DataAtom> {
+ @Override
+ public void exec(Tuple input, DataAtom output) throws IOException {
+ output.setValue(min(input));
+ }
+ }
+
+ static protected double min(Tuple input) throws IOException {
+ DataBag values = input.getBagField(0);
+
+ double curMin = Double.POSITIVE_INFINITY;
+ for (Iterator it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple) it.next();
+ try {
+ curMin = java.lang.Math.min(curMin, t.getAtomField(0).numval());
+ }catch(RuntimeException exp) {
+ IOException newE = new IOException("Error processing: " + t.toString() + exp.getMessage());
newE.initCause(exp);
throw newE;
- }
- }
+ }
+ }
- return curMin;
+ return curMin;
}
- @Override
- public Schema outputSchema(Schema input) {
- return new AtomSchema("min" + count++);
- }
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new AtomSchema("min" + count++);
+ }
- private static int count = 1;
+ private static int count = 1;
}
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/PigDump.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/PigDump.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/PigDump.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/PigDump.java Wed Feb 6 15:09:43 2008
@@ -26,21 +26,21 @@
public class PigDump implements StoreFunc {
- public static String recordDelimiter = "\n";
-
+ public static String recordDelimiter = "\n";
+
OutputStream os;
- public void bindTo(OutputStream os) throws IOException {
- this.os = os;
- }
+ public void bindTo(OutputStream os) throws IOException {
+ this.os = os;
+ }
- public void finish() throws IOException {
-
- }
+ public void finish() throws IOException {
+
+ }
- public void putNext(Tuple f) throws IOException {
- os.write((f.toString() + recordDelimiter).getBytes());
- }
+ public void putNext(Tuple f) throws IOException {
+ os.write((f.toString() + recordDelimiter).getBytes());
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/PigStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/PigStorage.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/PigStorage.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/PigStorage.java Wed Feb 6 15:09:43 2008
@@ -34,10 +34,10 @@
*/
public class PigStorage implements LoadFunc, StoreFunc {
protected BufferedPositionedInputStream in = null;
- long end = Long.MAX_VALUE;
- private byte recordDel = (byte)'\n';
- private String fieldDel = "\t";
- final private static Charset utf8 = Charset.forName("UTF8");
+ long end = Long.MAX_VALUE;
+ private byte recordDel = (byte)'\n';
+ private String fieldDel = "\t";
+ final private static Charset utf8 = Charset.forName("UTF8");
public PigStorage() {
}
@@ -54,7 +54,7 @@
this.fieldDel = delimiter;
}
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
if (in == null || in.getPosition() > end) {
return null;
}
@@ -65,7 +65,7 @@
return null;
}
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
+ public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
this.in = in;
this.end = end;
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/SUM.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/SUM.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/SUM.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/SUM.java Wed Feb 6 15:09:43 2008
@@ -53,13 +53,13 @@
static public class Initial extends EvalFunc<Tuple> {
@Override
- public void exec(Tuple input, Tuple output) throws IOException {
+ public void exec(Tuple input, Tuple output) throws IOException {
output.appendField(new DataAtom(sum(input)));
}
}
static public class Final extends EvalFunc<DataAtom> {
@Override
- public void exec(Tuple input, DataAtom output) throws IOException {
+ public void exec(Tuple input, DataAtom output) throws IOException {
output.setValue(sum(input));
}
}
@@ -68,18 +68,18 @@
DataBag values = input.getBagField(0);
double sum = 0;
- int i = 0;
+ int i = 0;
Tuple t = null;
for (Iterator it = values.iterator(); it.hasNext();) {
try {
t = (Tuple) it.next();
- i++;
+ i++;
sum += t.getAtomField(0).numval();
}catch(RuntimeException exp) {
- String msg = "iteration = " + i + "bag size = " + values.size() + " partial sum = " + sum + "\n";
- if (t != null)
- msg += "previous tupple = " + t.toString();
- throw new RuntimeException(exp.getMessage() + " additional info: " + msg);
+ String msg = "iteration = " + i + "bag size = " + values.size() + " partial sum = " + sum + "\n";
+ if (t != null)
+ msg += "previous tupple = " + t.toString();
+ throw new RuntimeException(exp.getMessage() + " additional info: " + msg);
//throw new RuntimeException(exp.getMessage() + " error processing: " + t.toString());
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/builtin/TextLoader.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/builtin/TextLoader.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/builtin/TextLoader.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/builtin/TextLoader.java Wed Feb 6 15:09:43 2008
@@ -31,11 +31,11 @@
* contains the line of text.
*/
public class TextLoader implements LoadFunc{
- BufferedPositionedInputStream in;
- final private static Charset utf8 = Charset.forName("UTF8");
- long end;
+ BufferedPositionedInputStream in;
+ final private static Charset utf8 = Charset.forName("UTF8");
+ long end;
- public void bindTo(String fileName, BufferedPositionedInputStream in, long offset, long end) throws IOException {
+ 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
@@ -44,7 +44,7 @@
getNext();
}
- public Tuple getNext() throws IOException {
+ public Tuple getNext() throws IOException {
if (in == null || in.getPosition() > end)
return null;
String line;