hazelcast-jet - Hazelcast Jet 0.6.1 + Pipeline API + 定制处理器
问题描述
我正在尝试将自定义处理器附加到 Hazelcast Jet Pipeline 定义。
这是示例代码。
private Pipeline buildPipeline() {
Pipeline p = Pipeline.create();
p.drawFrom(Sources.<String, Record>remoteMapJournal("record",
getClientConfig(), START_FROM_OLDEST))
.addTimestamps((v) -> getTimeStamp(v), 3000)
.peek()
.groupingKey((v) -> Tuple2.tuple2(getUserID(v),getTranType(v)))
.window(WindowDefinition.sliding(SLIDING_WINDOW_LENGTH_MILLIS,
SLIDE_STEP_MILLIS))
.aggregate(counting())
.map((v)-> getMapKey(v))
.customTransform("test2", ()-> this);
return p;
}
这是 tryProcess() 方法的示例代码
protected boolean tryProcess(int ordinal, Object item) {
TimestampedEntry entry = (TimestampedEntry)item;
System.out.println("Item value is "+ item);
map1.put(entry.getKey(), entry.getValue());
return true;
}
在执行时,我得到了以下异常
java.lang.IllegalArgumentException: These transforms have nothing attached to them: [test2]
at com.hazelcast.jet.impl.pipeline.Planner.validateNoLeakage(Planner.java:104)
at com.hazelcast.jet.impl.pipeline.Planner.createDag(Planner.java:65)
at com.hazelcast.jet.impl.pipeline.PipelineImpl.toDag(PipelineImpl.java:85)
at com.hazelcast.jet.JetInstance.newJob(JetInstance.java:94)
at com.sap.banking.bc.AbstractAuditLogProcessor1.start(AbstractAuditLogProcessor1.java:132)
at com.sap.banking.bc.JobProcessor.main(JobProcessor.java:18)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:282)
at java.lang.Thread.run(Thread.java:812)
有什么遗漏吗?你能帮我解决这个问题吗?
解决方案
例外是你错过了一个接收器,一个drainTo
电话。
推荐阅读
- scripting - ZSH 自动补全:显示值 A,使用值 B
- flutter - 如何在颤动中制作可扩展的卡片堆栈
- intellij-idea - IntelliJ 在使用 ZSH 时没有拾取环境变量?
- java - 我需要帮助让我的输入按钮在用鼠标单击时工作
- flutter - 如何在 Flutter 中构造表格数据
- python-3.x - UNET 训练:准确度从 0.99 开始
- c# - 使用 NPOI 从 Excel 中的单元格获取完整的日期和时间值
- r - 如何对来自多个数据框的数据进行分组
- tensorflow - 使用 model.fit 训练时列出超出范围的索引
- java - Android Studio 位置通知已更改服务