Apache Flink使用Java-性能问题

Question

我们有一个在 Java 上编写并在 AWS Kinesis Data Analytics 上运行的 flink 应用程序。应用程序从 AWS 托管服务 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是更明显的并行选择。(当然，如果每个槽也访问平面图中的外部服务，这也需要考虑在内。)

看到代码后更新：

你可以做几件简单的事情来加快速度。

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

但在这样做之前，请先看一下优化管道。重新平衡和keyBy都相当昂贵，并且您使用它们的次数超过了必要的程度。有一个键By，然后立即进行重新平衡是没有意义的，有两个keyBy的一个紧接着另一个也没有意义。

Rebalance对数据流进行往返重新分区。这通常是在改变并行度或需要克服数据不对称时进行的。几乎从不使用重新平衡，除非在需要更改并行度时隐式使用。

KeyBy 执行基于密钥的重新分区。如果一个键By跟随另一个键，则第二个键将撤消第一个键所做的任何操作。

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

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