我正在查看一些每秒生成约30,000条消息的kafka主题。我已经设置了一个flink拓扑结构来读取其中一个主题,进行一些聚合(5秒窗口),然后(最终)写入数据库。
当我运行我的拓扑结构并删除除读取->聚合步骤之外的所有内容时,我只能获得每分钟约30,000条消息。没有地方会出现背压(backpressure)。
我做错了什么?
编辑:
1. 我不能更改主题空间中的任何内容。每个主题只有一个分区,而且有数百个主题。 2. 每个消息是一个平均为2-3Kb的压缩thrift对象。
看起来我只能获得约1.5 MB/s。与提及的100MB/s相去甚远。
目前的代码路径:
当我运行我的拓扑结构并删除除读取->聚合步骤之外的所有内容时,我只能获得每分钟约30,000条消息。没有地方会出现背压(backpressure)。
我做错了什么?
编辑:
1. 我不能更改主题空间中的任何内容。每个主题只有一个分区,而且有数百个主题。 2. 每个消息是一个平均为2-3Kb的压缩thrift对象。
看起来我只能获得约1.5 MB/s。与提及的100MB/s相去甚远。
目前的代码路径:
DataStream<byte[]> dataStream4 = env.addSource(new FlinkKafkaConsumer081<>("data_4", new RawSchema(), parameterTool.getProperties())).setParallelism(1);
DataStream<Tuple4<Long, Long, Integer, String>> ds4 = dataStream4.rebalance().flatMap(new mapper2("data_4")).setParallelism(4);
public class mapper2 implements FlatMapFunction<byte[], Tuple4<Long, Long, Integer, String>> {
private String mapId;
public mapper2(String mapId) {
this.mapId = mapId;
}
@Override
public void flatMap(byte[] bytes, Collector<Tuple4<Long, Long, Integer, String>> collector) throws Exception {
TimeData timeData = (TimeData)ts_thriftDecoder.fromBytes(bytes);
Tuple4 tuple4 = new Tuple4<Long, Long, Integer, String>();
tuple4.f0 = timeData.getId();
tuple4.f1 = timeData.getOtherId();
tuple4.f2 = timeData.getSections().size();
tuple4.f3 = mapId;
collector.collect(tuple4);
}
}