[streaming] Ordered imports, migrated StringValues to IntValues and LongValues

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/DefaultPartitioner.java

 package eu.stratosphere.streaming;
 
-import eu.stratosphere.core.io.IOReadableWritable;
 import eu.stratosphere.nephele.io.ChannelSelector;
 import eu.stratosphere.types.Record;
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/JobGraphBuilder.java

@@ -13,65 +13,63 @@ import eu.stratosphere.nephele.jobgraph.JobTaskVertex;
 import eu.stratosphere.nephele.template.AbstractInputTask;
 import eu.stratosphere.nephele.template.AbstractOutputTask;
 import eu.stratosphere.nephele.template.AbstractTask;
- 
 
 public class JobGraphBuilder {
-	
+
 	private final JobGraph jobGraph;
-  private Map<String, AbstractJobVertex> components;
+	private Map<String, AbstractJobVertex> components;
 
-	
 	public JobGraphBuilder(String jobGraphName) {
-   
-	  jobGraph = new JobGraph(jobGraphName);
-	  components = new HashMap<String, AbstractJobVertex>();
-  }
-	
-	//TODO: Add source parallelism 
-	public void setSource(String sourceName, final Class<? extends AbstractInputTask<?>> sourceClass) {	  
-	  
-	  final JobInputVertex source = new JobInputVertex(sourceName, jobGraph);
-	  source.setInputClass(sourceClass);
-	  components.put(sourceName, source);
-	  
+
+		jobGraph = new JobGraph(jobGraphName);
+		components = new HashMap<String, AbstractJobVertex>();
 	}
-	
-	public void setTask(String taskName, final Class<? extends AbstractTask> taskClass, int parallelism) {   
-    
-    final JobTaskVertex task = new JobTaskVertex(taskName, jobGraph);
-    task.setTaskClass(taskClass);
-    task.setNumberOfSubtasks(parallelism);
-    components.put(taskName, task);
-  }
-	
-  public void setSink(String sinkName, final Class<? extends AbstractOutputTask> sinkClass) {    
-      
-      final JobOutputVertex sink = new JobOutputVertex(sinkName, jobGraph);
-      sink.setOutputClass(sinkClass);
-      components.put(sinkName, sink);
-    }
-  
-  
-  public void connect(String upStreamComponentName, String downStreamComponentName, ChannelType channelType) {
-    
-    AbstractJobVertex upStreamComponent=null;
-    AbstractJobVertex downStreamComponent=null;
-    
-    upStreamComponent = components.get(upStreamComponentName);
-    downStreamComponent = components.get(downStreamComponentName);
-    
-    try {
-      upStreamComponent.connectTo(downStreamComponent, channelType);
-    }
-    catch (JobGraphDefinitionException e) {
-      e.printStackTrace();
-    }
-}
 
+	// TODO: Add source parallelism
+	public void setSource(String sourceName,
+			final Class<? extends AbstractInputTask<?>> sourceClass) {
 
-public JobGraph getJobGraph() {
-  return jobGraph; 
-}
+		final JobInputVertex source = new JobInputVertex(sourceName, jobGraph);
+		source.setInputClass(sourceClass);
+		components.put(sourceName, source);
+
+	}
+
+	public void setTask(String taskName,
+			final Class<? extends AbstractTask> taskClass, int parallelism) {
+
+		final JobTaskVertex task = new JobTaskVertex(taskName, jobGraph);
+		task.setTaskClass(taskClass);
+		task.setNumberOfSubtasks(parallelism);
+		components.put(taskName, task);
+	}
 
+	public void setSink(String sinkName,
+			final Class<? extends AbstractOutputTask> sinkClass) {
+
+		final JobOutputVertex sink = new JobOutputVertex(sinkName, jobGraph);
+		sink.setOutputClass(sinkClass);
+		components.put(sinkName, sink);
+	}
+
+	public void connect(String upStreamComponentName,
+			String downStreamComponentName, ChannelType channelType) {
+
+		AbstractJobVertex upStreamComponent = null;
+		AbstractJobVertex downStreamComponent = null;
+
+		upStreamComponent = components.get(upStreamComponentName);
+		downStreamComponent = components.get(downStreamComponentName);
+
+		try {
+			upStreamComponent.connectTo(downStreamComponent, channelType);
+		} catch (JobGraphDefinitionException e) {
+			e.printStackTrace();
+		}
+	}
+
+	public JobGraph getJobGraph() {
+		return jobGraph;
+	}
 
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/MyStream.java

 package eu.stratosphere.streaming;
 
-import eu.stratosphere.streaming.cellinfo.WorkerEngineExact;
-
-import java.util.Random;
-
 import eu.stratosphere.configuration.Configuration;
-import eu.stratosphere.core.io.IOReadableWritable;
 import eu.stratosphere.core.io.StringRecord;
 import eu.stratosphere.nephele.io.ChannelSelector;
 import eu.stratosphere.nephele.io.RecordReader;
@@ -20,7 +15,10 @@ import eu.stratosphere.nephele.template.AbstractInputTask;
 import eu.stratosphere.nephele.template.AbstractOutputTask;
 import eu.stratosphere.nephele.template.AbstractTask;
 import eu.stratosphere.pact.runtime.task.util.TaskConfig;
+import eu.stratosphere.streaming.cellinfo.WorkerEngineExact;
 import eu.stratosphere.test.util.TestBase2;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
 
@@ -105,13 +103,22 @@ public class MyStream extends TestBase2 {
 
     @Override
     public void invoke() throws Exception {
-      Random rnd = new Random();
-
       for (int i = 0; i < 5; i++) {
-        // output.emit(new
-        // StringRecord(rnd.nextInt(10)+" "+rnd.nextInt(1000)+" 500"));
-        output.emit(new Record(new StringValue("5 510 100")));
-        output.emit(new Record(new StringValue("4 510 100")));
+      	Record record1 = new Record(3);
+      	record1.setField(0, new IntValue(5));
+      	record1.setField(1, new LongValue(510));
+      	record1.setField(2, new LongValue(100));
+      	
+      	Record record2 = new Record(3);
+      	record2.setField(0, new IntValue(4));
+      	record2.setField(1, new LongValue(510));
+      	record1.setField(2, new LongValue(100));
+      	
+      	output.emit(record1);
+      	output.emit(record2);
+      	
+//        output.emit(new Record(new StringValue("5 510 100")));
+//        output.emit(new Record(new StringValue("4 510 100")));
       }
 
     }
@@ -186,6 +193,16 @@ public class MyStream extends TestBase2 {
 
   }
 
+  private class QuerySourceInvokable implements UserSourceInvokable {
+		@Override
+		public void invoke(RecordWriter<Record> output) throws Exception {
+      for (int i = 0; i < 5; i++) {
+        output.emit(new Record(new StringValue("5 510 100")));
+        output.emit(new Record(new StringValue("4 510 100")));
+      }
+		}
+  }
+  
   @Override
   public JobGraph getJobGraph() {
 
@@ -196,6 +213,7 @@ public class MyStream extends TestBase2 {
     TaskConfig tConfig = new TaskConfig(infoSource.getConfiguration());
     Configuration config = tConfig.getConfiguration();
     config.setClass("partitioner", StreamPartitioner.class);
+    config.setClass("querySourceInvokable", QuerySourceInvokable.class);
     infoSource.setInputClass(InfoSource.class);
 
     final JobInputVertex querySource = new JobInputVertex("MyQuerySource", myJG);
@@ -217,7 +235,7 @@ public class MyStream extends TestBase2 {
       task1.connectTo(sink, ChannelType.INMEMORY);
 
     } catch (JobGraphDefinitionException e) {
-      // TODO Auto-generated catch block
+			// TODO Auto-generated catch block
       e.printStackTrace();
     }
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/StreamPartitioner.java

 package eu.stratosphere.streaming;
 
-import eu.stratosphere.core.io.StringRecord;
 import eu.stratosphere.nephele.io.ChannelSelector;
+import eu.stratosphere.types.IntValue;
 import eu.stratosphere.types.Record;
-import eu.stratosphere.types.StringValue;
 
 public class StreamPartitioner implements ChannelSelector<Record> {
 
-	/*@Override
-	public int[] selectChannels(StringRecord record, int numberOfOutputChannels) {
-		int cellId = Integer.parseInt(record.toString().split(" ")[0]);		
-		return new int[]{cellId % numberOfOutputChannels};
-	}*/
-
 	@Override
 	public int[] selectChannels(Record record, int numberOfOutputChannels) {
-		StringValue value = new StringValue("");
+		IntValue value = new IntValue();
 		record.getFieldInto(0, value);
-		int cellId = Integer.parseInt(value.getValue().split(" ")[0]);
+		int cellId = value.getValue();
 		return new int[]{cellId % numberOfOutputChannels};
 	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/StreamSource.java

 package eu.stratosphere.streaming;
 
-import eu.stratosphere.core.io.IOReadableWritable;
 import eu.stratosphere.nephele.io.ChannelSelector;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.nephele.template.AbstractInputTask;
@@ -8,60 +7,57 @@ import eu.stratosphere.types.Record;
 
 public class StreamSource extends AbstractInputTask<RandIS> {
 
-  private RecordWriter<Record> output;
-  private Class<? extends ChannelSelector<Record>> Partitioner;
-  ChannelSelector<Record> partitioner;
-  private Class<? extends UserSourceInvokable> UserFunction;
-  private UserSourceInvokable userFunction;
-  
-  public StreamSource() {
-    Partitioner = null;
-    UserFunction = null;
-    partitioner = null;
-    userFunction = null;
-  }
-
-  @Override
-  public RandIS[] computeInputSplits(int requestedMinNumber) throws Exception {
-    // TODO Auto-generated method stub
-    return null;
-  }
-
-  @Override
-  public Class<RandIS> getInputSplitType() {
-    // TODO Auto-generated method stub
-    return null;
-  }
-  
-  private void setClassInputs() {
-    Partitioner = getTaskConfiguration().getClass("partitioner", DefaultPartitioner.class,ChannelSelector.class);
-    try {
-      partitioner = Partitioner.newInstance();
-    } catch (Exception e) {
-
-    }
-    UserFunction = getTaskConfiguration().getClass("userfunction", TestSourceInvokable.class,UserSourceInvokable.class);
-    try
-    {
-      userFunction = UserFunction.newInstance();
-    } catch (Exception e)
-    {
-      
-    }
-  
-  }
-
-  @Override
-  public void registerInputOutput() {
-    setClassInputs();
-    output = new RecordWriter<Record>(this,
-        Record.class, this.partitioner);
-
-  }
-
-  @Override
-  public void invoke() throws Exception {
-    userFunction.invoke(output);
-  }
+	private RecordWriter<Record> output;
+	private Class<? extends ChannelSelector<Record>> Partitioner;
+	ChannelSelector<Record> partitioner;
+	private Class<? extends UserSourceInvokable> UserFunction;
+	private UserSourceInvokable userFunction;
+
+	public StreamSource() {
+		Partitioner = null;
+		UserFunction = null;
+		partitioner = null;
+		userFunction = null;
+	}
+
+	@Override
+	public RandIS[] computeInputSplits(int requestedMinNumber) throws Exception {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public Class<RandIS> getInputSplitType() {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	private void setClassInputs() {
+		Partitioner = getTaskConfiguration().getClass("partitioner",
+				DefaultPartitioner.class, ChannelSelector.class);
+		try {
+			partitioner = Partitioner.newInstance();
+		} catch (Exception e) {
+
+		}
+		UserFunction = getTaskConfiguration().getClass("userfunction",
+				TestSourceInvokable.class, UserSourceInvokable.class);
+		try {
+			userFunction = UserFunction.newInstance();
+		} catch (Exception e) {
+
+		}
+	}
+
+	@Override
+	public void registerInputOutput() {
+		setClassInputs();
+		output = new RecordWriter<Record>(this, Record.class, this.partitioner);
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		userFunction.invoke(output);
+	}
 
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/StreamTask.java

@@ -3,84 +3,74 @@ package eu.stratosphere.streaming;
 import java.util.ArrayList;
 import java.util.List;
 
-import eu.stratosphere.core.io.IOReadableWritable;
-import eu.stratosphere.core.io.StringRecord;
 import eu.stratosphere.nephele.io.ChannelSelector;
 import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.io.RecordWriter;
-import eu.stratosphere.nephele.template.AbstractInputTask;
 import eu.stratosphere.nephele.template.AbstractTask;
-import eu.stratosphere.streaming.cellinfo.WorkerEngineExact;
 import eu.stratosphere.types.Record;
 
 public class StreamTask extends AbstractTask {
 
-  private RecordWriter<Record> output;
-  private Class<? extends ChannelSelector<Record>> Partitioner;
-  ChannelSelector<Record> partitioner;
-  private Class<? extends UserTaskInvokable> UserFunction;
-  private UserTaskInvokable userFunction;
-  
-
-  private RecordReader<Record> inputInfo = null;
-  private RecordReader<Record> inputQuery = null;
-
-
-  
-  public StreamTask() {
-    Partitioner = null;
-    UserFunction = null;
-    partitioner = null;
-    userFunction = null;
-  }
-
-  
-  private void setClassInputs() {
-    Partitioner = getTaskConfiguration().getClass("partitioner", DefaultPartitioner.class,ChannelSelector.class);
-    try {
-      partitioner = Partitioner.newInstance();
-    } catch (Exception e) {
-
-    }
-    UserFunction = getTaskConfiguration().getClass("userfunction", TestTaskInvokable.class,UserTaskInvokable.class);
-    try
-    {
-      userFunction = UserFunction.newInstance();
-    } catch (Exception e)
-    {
-      
-    }
-  
-  }
-
-  @Override
-  public void registerInputOutput() {
-    setClassInputs();
-    this.inputInfo = new RecordReader<Record>(this, Record.class);
-    this.inputQuery = new RecordReader<Record>(this, Record.class);
-    output = new RecordWriter<Record>(this, Record.class, this.partitioner);
-    
-  }
-
-  @Override
-  public void invoke() throws Exception {
-    List< RecordReader<Record>> inputs = new ArrayList< RecordReader<Record>>();
-    inputs.add(inputInfo);
-    inputs.add(inputQuery);
-    
-    boolean hasInput = true;
-    while (hasInput)
-    {
-    	hasInput = false;
-    	for (RecordReader<Record> input : inputs)
-    	{
-    		if (input.hasNext()) {
-    			hasInput = true;
-    			userFunction.invoke(input.next(), output);
-    		}
-    	}
-    }
-
-  }
+	private RecordWriter<Record> output;
+	private Class<? extends ChannelSelector<Record>> Partitioner;
+	ChannelSelector<Record> partitioner;
+	private Class<? extends UserTaskInvokable> UserFunction;
+	private UserTaskInvokable userFunction;
+
+	private RecordReader<Record> inputInfo = null;
+	private RecordReader<Record> inputQuery = null;
+
+	public StreamTask() {
+		Partitioner = null;
+		UserFunction = null;
+		partitioner = null;
+		userFunction = null;
+	}
+
+	private void setClassInputs() {
+		Partitioner = getTaskConfiguration().getClass("partitioner",
+				DefaultPartitioner.class, ChannelSelector.class);
+		try {
+			partitioner = Partitioner.newInstance();
+		} catch (Exception e) {
+
+		}
+		UserFunction = getTaskConfiguration().getClass("userfunction",
+				TestTaskInvokable.class, UserTaskInvokable.class);
+		try {
+			userFunction = UserFunction.newInstance();
+		} catch (Exception e) {
+
+		}
+
+	}
+
+	@Override
+	public void registerInputOutput() {
+		setClassInputs();
+		this.inputInfo = new RecordReader<Record>(this, Record.class);
+		this.inputQuery = new RecordReader<Record>(this, Record.class);
+		output = new RecordWriter<Record>(this, Record.class, this.partitioner);
+
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		List<RecordReader<Record>> inputs = new ArrayList<RecordReader<Record>>();
+		inputs.add(inputInfo);
+		inputs.add(inputQuery);
+
+		boolean hasInput = true;
+		while (hasInput) {
+			hasInput = false;
+			for (RecordReader<Record> input : inputs) {
+				if (input.hasNext()) {
+					hasInput = true;
+					userFunction.invoke(input.next(), output);
+				}
+			}
+		}
+
+	}
 
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/TestSourceInvokable.java

 package eu.stratosphere.streaming;
 
 
-import eu.stratosphere.core.io.IOReadableWritable;
-import eu.stratosphere.core.io.StringRecord;
 import eu.stratosphere.nephele.io.RecordWriter;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
-import eu.stratosphere.types.StringValue;
-import eu.stratosphere.types.Value;
 
 public class TestSourceInvokable implements UserSourceInvokable {
 
   @Override
   public void invoke(RecordWriter<Record> output) throws Exception {
     for (int i = 0; i < 10; i++) {
-      // output.emit(new StringRecord(rnd.nextInt(10)+" "+rnd.nextInt(1000)));
-    	output.emit(new Record(new StringValue("5 500")));//new StringRecord("5 500"));
-    	output.emit(new Record(new StringValue("4 500")));
+    	Record record1 = new Record(2);
+    	record1.setField(0, new IntValue(5));
+    	record1.setField(1, new LongValue(500));
+    	Record record2 = new Record(2);
+    	record2.setField(0, new IntValue(4));
+    	record2.setField(1, new LongValue(500));
+    	output.emit(record1);
+    	output.emit(record2);
     }
   }
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/TestTaskInvokable.java

 package eu.stratosphere.streaming;
 
-import java.util.List;
-
-import eu.stratosphere.core.io.IOReadableWritable;
-import eu.stratosphere.core.io.StringRecord;
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.streaming.cellinfo.WorkerEngineExact;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.LongValue;
 import eu.stratosphere.types.Record;
 import eu.stratosphere.types.StringValue;
 
 public class TestTaskInvokable implements UserTaskInvokable {
 
-  private WorkerEngineExact engine = new WorkerEngineExact(10, 1000, 0);
+	private WorkerEngineExact engine = new WorkerEngineExact(10, 1000, 0);
 
-  @Override
-  public void invoke(Record record,
-      RecordWriter<Record> output) throws Exception {
-  		StringValue value = new StringValue();
-  		record.getFieldInto(0, value);
-  		String[] values = value.getValue().split(" ");
-  		
-  		//INFO
-  		if (values.length == 2)
-  		{
-  			engine.put(Integer.parseInt(values[0]), Long.parseLong(values[1]));
-  			output.emit(new Record(new StringValue(values[0] + " " + values[1])));
-  		}
-  		//QUERY
-  		else if (values.length == 3)
-  		{
-  			output.emit(new Record(new StringValue(String.valueOf(engine.get(
-          Long.parseLong(values[1]), Long.parseLong(values[2]),
-          Integer.parseInt(values[0]))))));
-  		}
-  		
-//  	RecordReader<IOReadableWritable> input1= inputs.get(0);
-//    RecordReader<IOReadableWritable> input2= inputs.get(0);
-//
-//    
-//    while (input1.hasNext() && input2.hasNext()) {
-//      String[] info = input1.next().toString().split(" ");
-//      String[] query = input2.next().toString().split(" ");
-//
-//      engine.put(Integer.parseInt(info[0]), Long.parseLong(info[1]));
-//
-//      output.emit(new StringRecord(info[0] + " " + info[1]));
-//      output.emit(new StringRecord(String.valueOf(engine.get(
-//          Long.parseLong(query[1]), Long.parseLong(query[2]),
-//          Integer.parseInt(query[0])))));
-//    }
-//    while (inputs.get(0).hasNext()) {
-//
-//      IOReadableWritable info = inputs.get(0).next();
-//
-//      output.emit(info);
-//    }
-//    while (inputs.get(1).hasNext()) {
-//
-//      IOReadableWritable query = inputs.get(1).next();
-//
-//      output.emit(query);
-//    }
-    
-  }
+	@Override
+	public void invoke(Record record, RecordWriter<Record> output)
+			throws Exception {
+		IntValue value1 = new IntValue(0);
+		record.getFieldInto(0, value1);
+		LongValue value2 = new LongValue(0);
+		record.getFieldInto(1, value2);
 
+		// INFO
+		if (record.getNumFields() == 2) {
+			engine.put(value1.getValue(), value2.getValue());
+			output.emit(new Record(new StringValue(value1 + " " + value2)));
+		}
+		// QUERY
+		else if (record.getNumFields() == 3) {
+			LongValue value3 = new LongValue(0);
+			record.getFieldInto(2, value3);
 
+			output.emit(new Record(new StringValue(String.valueOf(engine.get(
+					value2.getValue(), value3.getValue(), value1.getValue())))));
+		}
+	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/UserSourceInvokable.java

 package eu.stratosphere.streaming;
 
-import eu.stratosphere.core.io.IOReadableWritable;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.types.Record;
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/UserTaskInvokable.java

 package eu.stratosphere.streaming;
 
-import java.util.List;
-
-import eu.stratosphere.core.io.IOReadableWritable;
-import eu.stratosphere.nephele.io.RecordReader;
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.types.Record;
 
@@ -11,5 +7,4 @@ public interface UserTaskInvokable {
 
   public void invoke(Record record,
       RecordWriter<Record> output) throws Exception;
-
 }