Data stream Processing

StreamingJob.java

// set up the streaming execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// set time characteristic to be event timestamp
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// assign fictious stream of data as our data source and extract timestamp field
DataStream<Sensor> inputStream = env.addSource(new TemperatureSensor())
	.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Sensor>() {
	@Override
	public long extractAscendingTimestamp(Sensor sensor) {
		return sensor.getTimestamp();
	}
});

DataStream<Double> temps = inputStream.map(new MapFunction<Sensor, Double>() {
	@Override
	public Double map(Sensor sensor) throws Exception {
		return sensor.getTemperature();
	}
});