Hazelcast Jet-如何在Jet管道中使用非静态方法

问题描述

我在下面有一个基本的管道。在以下步骤之一中，我想通过从服务调用方法来转换对象，如下所示。但是Jet抛出错误，指出此mapFn不可序列化。在这里做什么？对于静态方法，它工作得很好。

p.readFrom(source)
     .map(r -> dataTransformer.transformRecord(r))// dataTransformer is a service
     .writeTo(Sinks.filesBuilder(userHome).build());

解决方法

使用mapUsingService并使用ServiceFactory创建服务：

p.readFrom(source)
 .mapUsingService(
    ServiceFactories.sharedService(pctx -> new DataTransformer()),(dataTransformer,r) -> dataTransformer.transformRecord(r))
 ...

或者，如果您的服务是可序列化的和无状态的，则可以将其复制到本地变量：

DataTransformer dataTransformerLocal = dataTransformer;
p.readFrom(source)
     .map(r -> dataTransformerLocal.transformRecord(r))
     .writeTo(Sinks.filesBuilder(userHome).build());