Hazelcast Jet Processor

前言

com.hazelcast.jet.core.processor.Processors（简称P）
这个类实现了核心的P，这里的P对应的是Jet引擎内部的DAG图的节点。
简单来说，这些P处理的是聚合（SUM/AVG/MAX/MIN），更复杂的情况是，可能是在任意的Key上做聚合，甚至是基于事件发生的时间窗上的聚合。因此Jet开发了两类P节点：

单阶段P
两阶段P
注意：从下面接口文档可以看出，无论是单阶段还是两阶段，都只适用于跑批的场景，并且重启job会丢失数据。
Jet内部默认传递给P的函数都是无状态的。

Jet的Streaming和batch

Jet区分steaming和batch非常简单，仅仅在P的complete方法上，返回false就是steaming的，返回true就是batch的。

单阶段聚合

所有的入站数据最好是预先分好区，groupby好，这样才不会在Jet集群中产生大量跨节点的数据传输。这种方式最省内存。

             -----------------| upstream vertex |-----------------|| partitioned-distributedV-----------| aggregate |-----------

两阶段聚合

第一个阶段只做accumulate，第二个阶段才做combine和finish动作。第一阶段完全在单节点本地，数据的分区，groupby只发生在第二阶段，因为第一阶段已经对数据做了聚合，到第二阶段数据传输量极小。

            -----------------| upstream vertex |-----------------|| partitioned-localV------------| accumulate |------------|| partitioned-distributedV----------------| combine/finish |----------------

在Jet中，单阶段聚合是两阶段聚合的特例

没有分区，groupby的情况下，第一阶段数据就在节点本地accumulate，然后第二阶段汇集到一个单一combine/finish节点上。

            -----------------| upstream vertex |-----------------|| local, non-partitionedV------------| accumulate |------------|| distributed, all-to-oneV----------------| combine/finish | localParallelism = 1----------------

Processors.java方法详解

public static <A,R> SupplierEx<Processor> aggregateP(@NonnullAggregateOperation<A,R> aggrOp)Type Parameters:
A - type of accumulator returned from aggrOp.createAccumulatorFn()
R - type of the finished result returned from aggrOp.finishAccumulationFn()
Parameters:
aggrOp - the aggregate operation to perform

消费入站数据，返回结果R，如果R是null，这个P就表示没有任何动作。
不写快照，重启任务会丢失数据，只适合做批处理。
——————————————————————————————————————

public static <A,R> SupplierEx<Processor> accumulateP(@NonnullAggregateOperation<A,R> aggrOp)

消费入站数据，返回结果A。
不写快照，重启任务会丢失数据，只适合做批处理。
——————————————————————————————————————

public static <A,R> SupplierEx<Processor> combineP(@NonnullAggregateOperation<A,R> aggrOp)

配合accumulateP使用，消费入站数据，返回结果R，如果R是null，这个P就表示没有任何动作。
不写快照，重启任务会丢失数据，只适合做批处理。
——————————————————————————————————————

public static <K,A,R,OUT> SupplierEx<Processor> aggregateByKeyP(List<FunctionEx<?,? extends K>> keyFns,AggregateOperation<A,R> aggrOp,BiFunctionEx<? super K,? super R,OUT> mapToOutputFn)
Type Parameters:
K - type of key
A - type of accumulator returned from aggrOp.createAccumulatorFn()
R - type of the result returned from aggrOp.finishAccumulationFn()
OUT - type of the item to emit
Parameters:
keyFns - functions that compute the grouping key
aggrOp - the aggregate operation
mapToOutputFn - function that takes the key and the aggregation result and returns the output item

可支持多路入站数据，每路的类型必须不同，每路要有自己的key抽取函数。aggOP里面的accumulate函数每路数据都要提供。
不写快照，重启任务会丢失数据，只适合做批处理。
————————————————————————————————————

public static <K,A> SupplierEx<Processor> accumulateByKeyP(List<FunctionEx<?,? extends K>> getKeyFns,AggregateOperation<A,?> aggrOp)

两阶段P的第一阶段P，根据分区，groupby应用accumulate，每个groupby的key输出一个Map.Entry<K, A>。
可支持多路入站数据，每路的类型必须不同，每路要有自己的key抽取函数。aggOP里面的accumulate函数每路数据都要提供。
不写快照，重启任务会丢失数据，只适合做批处理。
————————————————————————————————————

public static <K,A,R,OUT> SupplierEx<Processor> combineByKeyP(AggregateOperation<A,R> aggrOp,BiFunctionEx<? super K,? super R,OUT> mapToOutputFn)

两阶段P的第二阶段P，从多路accumulateByKeyP接收数据并应用combine，应用函数mapToOutputFn，每个key算出一个R。
不写快照，重启任务会丢失数据，只适合做批处理。
————————————————————————————————————

public static <K,A,R,OUT> SupplierEx<Processor> aggregateToSlidingWindowP(List<FunctionEx<?,? extends K>> keyFns,List<ToLongFunctionEx<?>> timestampFns,TimestampKind timestampKind,SlidingWindowPolicy winPolicy,long earlyResultsPeriod,AggregateOperation<A,? extends R> aggrOp,KeyedWindowResultFunction<? super K,? super R,? extends OUT> mapToOutputFn)

暂时用不到
————————————————————————————————————

public static <K,A> SupplierEx<Processor> accumulateByFrameP(List<FunctionEx<?,? extends K>> keyFns,List<ToLongFunctionEx<?>> timestampFns,TimestampKind timestampKind,SlidingWindowPolicy winPolicy,AggregateOperation<A,?> aggrOp)

暂时用不到
—————————————————————————————————————

public static <K,A,R,OUT> SupplierEx<Processor> combineToSlidingWindowP(SlidingWindowPolicy winPolicy,AggregateOperation<A,? extends R> aggrOp,KeyedWindowResultFunction<? super K,? super R,? extends OUT> mapToOutputFn)

暂时用不到
—————————————————————————————————————

public static <K,A,R,OUT> SupplierEx<Processor> aggregateToSessionWindowP(long sessionTimeout,long earlyResultsPeriod,List<ToLongFunctionEx<?>> timestampFns,List<FunctionEx<?,? extends K>> keyFns,AggregateOperation<A,? extends R> aggrOp,KeyedWindowResultFunction<? super K,? super R,? extends OUT> mapToOutputFn)

暂时用不到
————————————————————————————————————

public static <T> SupplierEx<Processor> insertWatermarksP(@NonnullEventTimePolicy<? super T> eventTimePolicy)

流处理的API来了！！！
在流里面插入watermark，wartermark由eventTimePolicy给出。
这个P会丢弃迟到的（late）数据。
最晚emit水印的数据值会被写入快照。重启所有节点，会从快照继续往后执行。
听起来视乎破坏了单调性原则，但是Jet会合并watermark，可以保证从快照继续往后执行。副作用就是，重启前被认为是迟到的（late）的event，重启后不会被认为是迟到的了。
——————————————————————————————————————

public static <T,R> SupplierEx<Processor> mapP(@NonnullFunctionEx<? super T,? extends R> mapFn)
Type Parameters:
T - type of received item
R - type of emitted item
Parameters:
mapFn - a stateless mapping function

这个函数是无状态的，效果等同于java流式计算中的filter
————————————————————————————————————————

public static <T> SupplierEx<Processor> filterP(@NonnullPredicateEx<? super T> filterFn)

这个函数是无状态的，根据Predicate条件过滤数据。
————————————————————————————————————————

public static <T,R> SupplierEx<Processor> flatMapP(FunctionEx<? super T,? extends Traverser<? extends R>> flatMapFn)

这个函数是无状态的，把数据转换成Traverser。
Traverser不能为null，可以是empty。
————————————————————————————————————————

public static <T,K,S,R> SupplierEx<Processor> mapStatefulP(long ttl,FunctionEx<? super T,? extends K> keyFn,ToLongFunctionEx<? super T> timestampFn,Supplier<? extends S> createFn,TriFunction<? super S,? super K,? super T,? extends R> statefulMapFn,TriFunction<? super S,? super K,? super Long,? extends R> onEvictFn)

有状态的映射，createFn函数返回状态对象s，每个group key就有一个对应的对象s，s对象会有快照，重启不会丢。所以对象s必须可序列化。如果MapFn把输入映射为了null，相当于过滤掉了这个输入数据。对象s的timestamp和watermark两个时间参数（watermark - ttl ？？？ timestamp）可以控制对象s是否要被丢失或者用watermark更新timestamp。
————————————————————————————————

public static <T,K,S,R> SupplierEx<Processor> flatMapStatefulP(long ttl,FunctionEx<? super T,? extends K> keyFn,ToLongFunctionEx<? super T> timestampFn,Supplier<? extends S> createFn,TriFunction<? super S,? super K,? super T,? extends Traverser<R>> statefulFlatMapFn,TriFunction<? super S,? super K,? super Long,? extends Traverser<R>> onEvictFn)

有状态的映射，createFn函数返回状态对象s，每个group key就有一个对应的对象s，s对象会有快照，重启不会丢。所以对象s必须可序列化。对象s的timestamp和watermark两个时间参数（watermark - ttl ？？？ timestamp）可以控制对象s是否要被丢失或者用watermark更新timestamp。
注意这个函数跟上一个函数的区别，请跟无状态的两个对应函数对比。
————————————————————————————————————

public static <C,S,T,R> ProcessorSupplier mapUsingServiceP(ServiceFactory<C,S> serviceFactory,BiFunctionEx<? super S,? super T,? extends R> mapFn)
Type Parameters:
C - type of context object
S - type of service object
T - type of received item
R - type of emitted item

对每个输入数据应用无状态函数mapFn,其中参数S是serviceFactory生成的。
————————————————————————————————————

public static <C,S,T,K,R> ProcessorSupplier mapUsingServiceAsyncP(ServiceFactory<C,S> serviceFactory,int maxConcurrentOps,boolean preserveOrder,FunctionEx<T,K> extractKeyFn,BiFunctionEx<? super S,? super T,CompletableFuture<R>> mapAsyncFn)
Type Parameters:
C - type of context object
S - type of service object
T - type of received item
K - type of key
R - type of result item
Parameters:
serviceFactory - the service factory
maxConcurrentOps - maximum number of concurrent async operations per processor
preserveOrder - whether the async responses are ordered or not
extractKeyFn - a function to extract snapshot keys. Used only if preserveOrder==false
mapAsyncFn - a stateless mapping function

future可以为null，future不为null它里面的结果也可以为null，如果是null代码此数据被过滤掉了。extractKeyFn是用来生产快照key的。
快照key的限制：如果接收到的数据跨越了分片，需要保证key相同，如果是round-robin那就随意了，可以用Object::hashCode
——————————————————————————————————

public static <C,S,T> ProcessorSupplier filterUsingServiceP(ServiceFactory<C,S> serviceFactory,BiPredicateEx<? super S,? super T> filterFn)

serviceFactory生成一个S，然后传递给Predicate函数，过滤数据。
————————————————————————————————————

public static <C,S,T,R> ProcessorSupplier flatMapUsingServiceP(ServiceFactory<C,S> serviceFactory,BiFunctionEx<? super S,? super T,? extends Traverser<R>> flatMapFn)

把输入数据转换成Traverser对象，Traverser对象结尾不为null，serviceFactory生成一个S，传递给flatMapFn。
转换过程中有S对象可以用本地存储，所以虽然有快照，但是不能保证一致性，可能出异常情况。
————————————————————————————————————

public static <T> SupplierEx<Processor> sortP(Comparator<T> comparator)

把输入数据按PriorityQueue顺序排序，并在complete阶段发射出去。
————————————————————————————————————

public static SupplierEx<Processor> noopP()

不用解释了吧，有进没出的P

无状态transforms

没有中间状态存储，所有item跟其他item无关。比如下面的算子：
map
filter
flatMap
merge
mapUsingIMap
mapUsingRelicateMap
mapUsingService
mapUsingServiceAsync
mapUsingServiceAsyncBatched
mapUsingPython
hashJoin

有状态transforms

有状态的transforms累计数据，因此输出结果依赖之前的item。
在Jet中窗口只能用于流式数据（带时间），一次性的聚合操作只适合batch。
aggregate
聚合是分布式流式数据处理的基石。它对流式的item数据执行sum或者avg等算子。AggregateOperations提供了一系列的聚合方法，比如：

averagingLong()
averageDouble()
counting()
summingLong()
summingDouble()
maxBy()
minBy()
toList()
bottomN()
topN()
linearTrend()
allOf()

groupingKey
略
rollingAggregate
不等接收所有的item，每接收一个item就输出一个结果。所以它可以在流式处理中使用。
window
把一个无界的数据流分成一块块。
tumblingWindow
略
slidingWindow
略
sessionWindow
略
Early Results
略
distinct
略过重复
sort
只支持batch
mapStateful
是map transform的扩展，它可以保存一个可变的中间状态。可以根据这个基础算子写出任意复杂的检查模式的状态机。
比如有个流中有TRANSACTION_START和TRANSACTION_END两个item数据。这种情况用window不行，因为不确定window大小。sessionWindow也不行，因为它默认是等待到最后一次性emit所有items。

public class TransactionEvent {long timestamp();String transactionId();EventType type();
}public enum EventType {TRANSACTION_START,TRANSACTION_END
}

可以这么写代码：

p.readFrom(KafkaSources.kafka(.., "transaction-events")).withNativeTimestamps(0).groupingKey(event -> event.getTransactionId()).mapStateful(MINUTES.toMillis(10),() -> new TransactionEvent[2],(state, id, event) -> {if (event.type() == TRANSACTION_START) {state[0] = event;} else if (event.type() == TRANSACTION_END) {state[1] = event;}if (state[0] != null && state[1] != null) {// we have both start and end eventslong duration = state[1].timestamp() - state[0].timestamp();return MapUtil.entry(event.transactionId(), duration);}// we don't have both events, do nothing for now.return null;},(state, id, currentWatermark) ->// if we have not received both events after 10 minutes,// we will emit a timeout entry(state[0] == null || state[1] == null)? MapUtil.entry(id, TIMED_OUT): null).writeTo(Sinks.logger());