1. ProcessFunctions

ProcessFunction能夠同時(shí)處理timers和state盗似，這也是創(chuàng)建event-driven application的基礎(chǔ)挟秤。

// compute the sum of the tips per hour for each driver
DataStream<Tuple3<Long, Long, Float>> hourlyTips = fares
        .keyBy((TaxiFare fare) -> fare.driverId)
        .process(new PseudoWindow(Time.hours(1)));

hourlyTips.getSideOutput(lateFares).print();

// Compute the sum of the tips for each driver in hour-long windows.
// The keys are driverIds.
public static class PseudoWindow extends 
        KeyedProcessFunction<Long, TaxiFare, Tuple3<Long, Long, Float>> {

    private final long durationMsec;
    private static final OutputTag<TaxiFare> lateFares = new OutputTag<TaxiFare>("lateFares") {};

    public PseudoWindow(Time duration) {
        this.durationMsec = duration.toMilliseconds();
    }

    // Keyed, managed state, with an entry for each window, keyed by the window's end time.
    // There is a separate MapState object for each driver.
    private transient MapState<Long, Float> sumOfTips;

    @Override
    // Called once during initialization.
    public void open(Configuration conf) {
        MapStateDescriptor<Long, Float> sumDesc =
        new MapStateDescriptor<>("sumOfTips", Long.class, Float.class);
        sumOfTips = getRuntimeContext().getMapState(sumDesc);
    }

    @Override
    // Called as each fare arrives to be processed.
    public void processElement(
            TaxiFare fare,
            Context ctx,
            Collector<Tuple3<Long, Long, Float>> out) throws Exception {
        long eventTime = fare.getEventTime();
        TimerService timerService = ctx.timerService();
        
        if (eventTime <= timerService.currentWatermark()) {
            // This event is late; its window has already been triggered.
            ctx.output(lateFares, fare);
        } else {
            // Round up eventTime to the end of the window containing this event.
            long endOfWindow = (eventTime - (eventTime % durationMsec) +     durationMsec - 1);

            // Schedule a callback for when the window has been completed.
            timerService.registerEventTimeTimer(endOfWindow);

            // Add this fare's tip to the running total for that window.
            Float sum = sumOfTips.get(endOfWindow);
            if (sum == null) {
                sum = 0.0F;
            }
            sum += fare.tip;
            sumOfTips.put(endOfWindow, sum);
        }
    }

    @Override
    // Called when the current watermark indicates that a window is now complete.
    public void onTimer(long timestamp, 
            OnTimerContext context, 
            Collector<Tuple3<Long, Long, Float>> out) throws Exception {

        long driverId = context.getCurrentKey();
        // Look up the result for the hour that just ended.
        Float sumOfTips = this.sumOfTips.get(timestamp);

        Tuple3<Long, Long, Float> result = Tuple3.of(driverId, timestamp, sumOfTips);
        out.collect(result);
        this.sumOfTips.remove(timestamp);
    }
}

性能考量：
Flink的MapState和ListState是為RocksDB做了優(yōu)化的阎肝。RocksDB可以向ListState追加元素而不需要重新序列化或反序列化渐北。而MapState筛璧，每一個(gè)key-value都是一個(gè)RocksDB對象牛欢，所以可以高效地訪問或更新骡男。