Flink kafka消费者落后

Question

我在Flink的工作中使用Kafka资料来源的信息流，一次阅读50个主题，如下所示：

FlinkKafkaConsumer<GenericRecord> kafkaConsumer = new FlinkKafkaConsumer<GenericRecord>(
            Pattern.compile("TOPIC_NAME[1-50].stream"), // getting data stream from all topics
            <DeserializationSchema>, //using avro schema
            properties); // auto.commit.interval.ms=1000 ...

然后有一些运算符，如：过滤器-

我能获得的最大吞吐量是每秒10k到20k条记录，考虑到源发布了数十万个事件，这相当低，我可以清楚地看到消费者落后于生产者。我甚至试着移除水槽和其他操作员，以确保没有背压，但它仍然是一样的。我正在将我的应用程序部署到Amazon Kinesis data analytics，并尝试了几种并行设置，但这些设置似乎都无法提高吞吐量。

我遗漏了什么吗？

Answer 1

有几件事会显着影响吞吐量。

无效的序列化通常是导致吞吐量低的一个主要因素。请参阅Flink Serialization Tuning Vol.1：选择您的序列化程序-如果可以的话，以获取有关此主题的详细信息。Avro通用记录序列化程序还不错，但您是否携带了实际不需要的数据？

您是否正在更改管道中的任何位置的并行性？那很昂贵。

对于Kinesis数据分析，您必须使用RocksDB状态后端，它的吞吐量远远低于基于堆的状态后端。但拥有正确的配置会有很大帮助。您应该为RocksDB工作目录使用可用的最快本地磁盘（SSD，或者在极端情况下，可能需要RAM磁盘）。确保实例类型提供足够的IOPs。给RocksDB足够的内存。如果您进行大量读取，那么Bloom过滤器值得启用。请参阅Flink中磁盘对RocksDB状态后端的影响：案例研究，以了解有关使用RocksDB的更多信息。

您可以尝试禁用检查点作为实验。如果这有帮助，这将提供一些线索。

某些网络设置会影响吞吐量。默认值通常提供了良好的性能，但如果您对其进行了修改，则值得研究。