我想对Flink CEP引擎进行性能分析，我遇到了这些类
org.apache.flink.optimizer.costs.CostEstimator;
org.apache.flink.optimizer.costs.Costs;
org.apache.flink.optimizer.costs.DefaultCostEstimator;
但问题是我不知道如何使用这个类 . 有人可以向我提供代码或暗示，如何在Flink中找到运营商{加入例如}的成本估算 .
下面是我在Flink中执行的联接的代码
DataStream<JoinedEvent> joinedEventDataStream = stream1.join(stream2).where(new KeySelector<RRIntervalStreamEvent, Long>() {
@Override
public Long getKey(RRIntervalStreamEvent rrIntervalStreamEvent) throws Exception {
return rrIntervalStreamEvent.getTime();
}
})
.equalTo(new KeySelector<qrsIntervalStreamEvent, Long>() {
@Override
public Long getKey(qrsIntervalStreamEvent qrsIntervalStreamEvent) throws Exception {
return qrsIntervalStreamEvent.getTime();
}
})
.window(TumblingEventTimeWindows.of(Time.milliseconds(1000)))
.apply(new JoinFunction<RRIntervalStreamEvent, qrsIntervalStreamEvent, JoinedEvent>() {
@Override
public JoinedEvent join(RRIntervalStreamEvent rr, qrsIntervalStreamEvent qrs) throws Exception {
//getting the cost -- just checking
// costs.getCpuCost();
return new JoinedEvent(rr.getTime(),rr.getSensor_id(),qrs.getSensor_id(),rr.getRRInterval(),qrs.getQrsInterval());
}
});
如何计算此联接的费用？
1 回答
成本类属于DataSet API的优化器（Flink的批处理API），而CEP库是基于DataStream API构建的 . DataStream API不利用DataSet API .
CEP库和DataSet优化器完全不相关 . 因此，不可能使用此代码来估计CEP模式的成本 . 我也没有意识到另一种估计CEP模式（或任何其他DataStream程序）成本的内置方法 .