wordcount 测试显示 Flink 速度缓慢 经过测试的 WordCount 管道

问题描述

我正在做一些流处理框架之间的基准比较，

我在这方面选择了 WordCount 这样的“Hello world”任务（有一些曲折），到目前为止测试了 Flink 和 Hazelcast Jet，结果是 Flink 需要 80+s 才能完成，而 Jet 只需要 30+s

我知道 Flink 很受欢迎，我在这里做错了什么？真的很好奇这个

我的示例代码在这里

https://github.com/ChinW/stream-processing-compare

---

以下是详细信息（规范、管道、日志）

经过测试的 WordCount 管道

Source (read from file,5MB)
 -> Process: Split line into words (Here here is a bomb,every word emit 1000 times)
 -> Group/Count
 -> Sink (do nothing)

我的本地结果

• MacBook Pro（13 英寸，2020 年，四个雷雳 3 端口）
• 2 GHz 四核 Intel Core i5（8 个逻辑处理器）
• 16 GB 3733 MHz LPDDR4X
• JDK 11

喷射 4.4

管道：

digraph DAG {
    "items" [localParallelism=1];
    "fused(flat-map,filter)" [localParallelism=8];
    "group-and-aggregate-prepare" [localParallelism=8];
    "group-and-aggregate" [localParallelism=8];
    "do-nothing-sink" [localParallelism=1];
    "items" -> "fused(flat-map,filter)" [queueSize=1024];
    "fused(flat-map,filter)" -> "group-and-aggregate-prepare" [label="partitioned",queueSize=1024];
    subgraph cluster_0 {
        "group-and-aggregate-prepare" -> "group-and-aggregate" [label="distributed-partitioned",queueSize=1024];
    }
    "group-and-aggregate" -> "do-nothing-sink" [queueSize=1024];
}

日志：

Start time: 2021-04-18T13:52:52.106
Duration: 00:00:36.459
Jet: finish in 36.45935081 seconds.

Start time: 2021-04-19T16:51:53.806
Duration: 00:00:30.143
Jet: finish in 30.625740453 seconds.

Start time: 2021-04-19T16:52:48.906
Duration: 00:00:37.207
Jet: finish in 37.862554137 seconds.

Scala 2.11 的 Flink 1.12.2

flink-config.yaml 配置：

jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 2096m
taskmanager.memory.process.size: 12288m
taskmanager.numberOfTaskSlots: 8
parallelism.default: 8

管道：

{
  "nodes" : [ {
    "id" : 1,"type" : "Source: Custom Source","pact" : "Data Source","contents" : "Source: Custom Source","parallelism" : 1
  },{
    "id" : 2,"type" : "Flat Map","pact" : "Operator","contents" : "Flat Map","parallelism" : 8,"predecessors" : [ {
      "id" : 1,"ship_strategy" : "REBALANCE","side" : "second"
    } ]
  },{
    "id" : 4,"type" : "Keyed Aggregation","contents" : "Keyed Aggregation","predecessors" : [ {
      "id" : 2,"ship_strategy" : "HASH",{
    "id" : 5,"type" : "Sink: Unnamed","pact" : "Data Sink","contents" : "Sink: Unnamed","predecessors" : [ {
      "id" : 4,"ship_strategy" : "FORWARD","side" : "second"
    } ]
  } ]
}

日志：

❯ flink run -c chiw.spc.flink.FlinkWordCountKt stream-processing-compare-1.0-SNAPSHOT.jar
Job has been submitted with JobID 163ce849a663e45f3c3028a98f260e7c
Program execution finished
Job with JobID 163ce849a663e45f3c3028a98f260e7c has finished.
Job Runtime: 88614 ms

❯ flink run -c chiw.spc.flink.FlinkWordCountKt stream-processing-compare-1.0-SNAPSHOT.jar
Job has been submitted with JobID fcf12488204969299e4e5d7f23f4ea6e
Program execution finished
Job with JobID fcf12488204969299e4e5d7f23f4ea6e has finished.
Job Runtime: 90165 ms

❯ flink run -c chiw.spc.flink.FlinkWordCountKt stream-processing-compare-1.0-SNAPSHOT.jar
Job has been submitted with JobID 37e349e4fad90cd7405546d30239afa4
Program execution finished
Job with JobID 37e349e4fad90cd7405546d30239afa4 has finished.
Job Runtime: 78908 ms

非常感谢您的帮助！

解决方法

我认为您没有做错任何事情，我们的测试表明 Jet 比 Spark 和 Flink 快得多，而字数统计是我们用来衡量这一点的示例之一。

,

鉴于您的炸弹会产生大量的小物品（而不是少量的大物品），我最好的猜测是为什么 Jet 在这里可能具有优势是它的单一生产者- 单消费者 (SPSC) 队列与类似协程的并发性相结合。

您有 8 个平面映射阶段和 8 个聚合阶段。 Jet 将在总共 8 个线程上执行此操作（假设您有 8 个 availableProcessors），因此几乎不会在操作系统级别进行线程调度。数据将以大块的形式在线程之间移动：flatMap 将一次排队 1024 个，然后每个聚合器将拉出所有指定给它的项目。 SPSC 队列上的通信不受其他线程的任何干扰：每个聚合处理器有 8 个输入队列，一个专用于每个平面映射器。

在 Flink 中，每个阶段将启动另外 8 个线程，我还注意到接收器的并行度为 8，因此这是 24 个线程，另一个用于源。操作系统必须将它们安排在 8 个物理内核上。通信将通过多生产者单消费者 (MPSC) 队列进行，这意味着所有平面映射器线程必须协调，以便一次只有一个线程将一个项目排队到任何给定的聚合器，并且争用导致热 CAS 循环所有线程。

要确认这种怀疑，请尝试收集一些分析数据。如果上面的故事是正确的，你应该看到 Flink 花费了大量的 CPU 时间来排队数据。