使用 Java 的 Apache Flink - 性能问题

Question

我们有一个使用 Java 编写并在 AWS Kinesis Data Analytics 上运行的 flink 应用程序。应用程序从 AWS Managed Service Kafka（kafka 主题 1）读取输入流，然后应用业务逻辑（一些计算），最后将输出写入另一个 Kafka 主题（kafka 主题 2）。

并行度为 10，主题有 15 个分区。 期望在 5 分钟内处理约 20K 并发数据。但经过所有优化后，我们可以在 25 分钟内将其提高到约 20K 并发数据的速度。

能否请您告诉我是否可以实施任何其他性能优化来实现目标。

Flink Async I/O 是否会成为进一步优化的选项？

示例代码：-

StreamExecutionEnvironment streamenv =
    StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<ObjectNode> initialStreamData = streamenv
    .addSource(new FlinkKafkaConsumer<>(
        TOPIC_NAME, 
        new ObjectNodeJsonDeSerializerSchema(),
        kafkaConnectProperties);

initialStreamData.print();

DataStream<POJO> rawDataProcess = initialStreamData
    .rebalance()
    .flatMap(new ReProcessingDataProvider())
    .keyBy(value -> value.getPersonId());

rawDataProcess.print();

DataStream<POJO> cgmStream = rawDataProcess
    .keyBy(new ReProcessorKeySelector())
    .rebalance()
    .flatMap(new SgStreamTask());

cgmStream.print();

DataStream<POJO> artfctOverlapStream = null;
artfctOverlapStream = cgmStreamData
    .keyBy(new CGMKeySelector())
    .countWindow(2, 1)
    .apply(new ArtifactOverlapProvider()); //the same person_id key

cgmStreamData.print();

DataStream<POJO> streamWithSgRoc = null;

streamWithSgRoc = artfctOverlapStream
    .keyBy(new CGMKeySelector())
    .countWindow(7, 1)
    .apply(new SgRocProvider()); // the same person_id key 

streamWithSgRoc.print();

DataStream<POJO> cgmExcursionStream = null;

cgmExcursionStream = streamWithSgRoc
    .keyBy(new CGMKeySelector())
    .countWindow(Common.THREE, Common.ONE)
    .apply(new CGMExcursionProviderStream()); //the same person_id key

cgmExcursionStream.print();

cgmExcursionStream
    .addSink(new FlinkKafkaProducer<CGMDataCollector>(
        topicsProperties.getProperty(Common.CGM_EVENT_TOPIC),
        new CGMDataCollectorSchema(),
        kafkaConnectProperties));

Answer 1

在您分享的内容中，我没有看到任何可以解释吞吐量不佳的内容。但既然你问过异步 i/o，我想知道平面图是否在做一些外部 i/o。如果是这样，那就可以解释了。如果是这种情况，那么使用异步 i/o 应该会有很大帮助，前提是外部服务可以处理增加的负载。

我也想知道为什么并行度是 10，以及这 10 个插槽有哪些资源可用。是否有足够的核心来保持一切运转？对于 15 个分区，您有 5 个插槽，每个插槽处理两个分区，另外 5 个插槽每个处理一个分区。 5、8 和 15 是更明显的并行选择。 （当然，如果每个 slot 也命中了 flatmap 中的外部服务，那也需要考虑在内。）

看到代码后更新：

您可以做一些简单的事情来加快速度。

要做的一件事是给集群更多的资源。您可以保持并行度不变，但通过将任务管理器放置在具有更多内核的机器上，为每个插槽提供更多可使用的内核。

但在此之前，请先了解一下优化管道。 rebalance 和 keyBy 都相当昂贵，而且您使用它们的次数超出了必要的范围。一个 keyBy 紧跟一个再平衡是没有意义的，两个 keyBy 一个接一个也没有意义。

Rebalance 对流进行往返重新分区。它通常在更改并行性或需要克服数据倾斜时完成。几乎从不使用重新平衡，除非在更改并行度时根据需要隐式使用。

KeyBy 执行基于键的重新分区。如果一个 keyBy 跟随另一个，第二个会撤消第一个所做的。

keyBy 和 rebalance 都需要序列化和反序列化每个事件，并通过网络堆栈发送它们。这是您仅在绝对必要时才想做的事情。

修复这些重新平衡/keyBy 问题将减少集群上的工作负载。如果这还不足以实现所需的吞吐量，那么为每个插槽提供更多内核（以便管道的各个阶段可以并行运行）应该可以解决问题。