is called with the key of the window, an Iterable to access the elements of the window, and a Collector to emit results. • A Context gives access to the metadata of the window (start and end timestamps // Evaluates the window void process( KEY key, Context ctx, Iterable vals, Collector out) throws Exception; public abstract class Context implements Serializable { processElement(v: IN, ctx: Context, out: Collector[OUT]) is called for each record of the stream. Result records are emitted by passing them to the Collector. The Context object gives access to the timestamp

FlatMapFunction { @Override public void flatMap(TaxiRide taxiRide, Collector out) throws Exception { FilterFunction valid = new RideCleansing.NYCFilter(); Minutes>>() { @Override public void flatMap(EnrichedRide ride, Collector> out) throws Exception { if (!ride.isStart) { Interval flatMap1(String control_value, Collector out) throws Exception { blocked.update(Boolean.TRUE); } @Override public void flatMap2(String data_value, Collector out) throws Exception

orReading, (String, Double, Double)] { … override def flatMap( reading: SensorReading, out: Collector[(String, Double, Double)]): Unit = { // fetch the last temperature from state val lastTemp = lastTempState ("saved fare", TaxiFare.class));
 }
 
 @Override
 public void flatMap1(TaxiRide ride, Collector> out) throws Exception {
 TaxiFare fare = fareState.value();
 matching fare -> store the ride
 }
 }
 
 @Override
 public void flatMap2(TaxiFare fare, Collector> out) throws Exception {
 // similar logic for processing fare

Double), out: Collector[Offer])= { factorValues.put(value._1, value._2) } // flatMap method for the stream of price requests override def flatMap2(item: Item, out: Collector[Offer]) = {