// create a new MovingAverage (with window size 2) if none exists for this key if (average == null) average = new MovingAverage(2); // add this event to the moving average average.add(item flatMap2(String data_value, Collector out) throws Exception { if (blocked.value() == null) { out.collect(data_value); } } } Lab 2 - Stateful Enrichment of Rides and Fares not-yet-fully-sorted events */ private ValueState> queueState = null; @Override public void open(Configuration config) { /* set up the state we want to use

produced till step k. 27 Vasiliki Kalavri | Boston University 2020 A null operator N is one where N(S) = [ ] for every S. A non-null operator G is • blocking, when for every sequence S of length n, operator is null when it returns the empty sequence for every possible value of its argument(s). A non-null unary operator G is non-blocking, when Gτ (S) = G(Sτ), for every τ. A non-null binary operator

reading is from sensor_1 new Watermark(extractedTS - bound) } else { // do not emit a watermark null } } override def extractTimestamp(r: Reading, prevTS: Long): Long = { // assign record timestamp

de, TaxiFare>> out) throws Exception {
 TaxiFare fare = fareState.value();
 if (fare != null) { // a matching fare exists
 fareState.clear(); // always clear the state you don’t need anymore

read from the socket connection val userInput = reader.readLine() while(! isStopped && userInput ! = null) { store(userInput) userInput = reader.readLine() } ... }} 17 / 79 Input Operations - Custom