Тест подсчета слов показывает медлительность Flink

я делаю некоторое сравнительное сравнение между платформами потоковой обработки,

Я выбрал WordCount, такую ​​​​задачу Hello world (с некоторыми поворотами) в этой области, и до сих пор тестировал Flink и Hazelcast Jet, в результате Flink требуется 80+ с, а Jet - только 30+.

Я знаю, что 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)
My Local Result
  • MacBook Pro (13 дюймов, 2020 г., четыре порта Thunderbolt 3)
  • Четырехъядерный процессор Intel Core i5 с тактовой частотой 2 ГГц (8 логических процессоров)
  • 16 ГБ 3733 МГц LPDDR4X
  • JDK 11
Jet 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.
Flink 1.12.2 for Scala 2.11

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",
    "pact" : "Operator",
    "contents" : "Keyed Aggregation",
    "parallelism" : 8,
    "predecessors" : [ {
      "id" : 2,
      "ship_strategy" : "HASH",
      "side" : "second"
    } ]
  }, {
    "id" : 5,
    "type" : "Sink: Unnamed",
    "pact" : "Data Sink",
    "contents" : "Sink: Unnamed",
    "parallelism" : 8,
    "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

Большое спасибо за помощь!


person Kuawiiii    schedule 19.04.2021    source источник


Ответы (2)


Я не думаю, что вы сделали что-то неправильно, наше тестирование показало, что Jet значительно быстрее, чем Spark и Flink, и количество слов — один из примеров, который мы использовали для измерения этого.

person Mike Yawn    schedule 19.04.2021

Учитывая, что ваша бомба создает большое количество мелких предметов (в отличие от меньшего количества больших предметов), я могу предположить, почему Jet может иметь здесь преимущество, так это его единственный производитель-единственный- потребительские (SPSC) очереди в сочетании с параллелизмом, подобным сопрограмме.

У вас есть 8 этапов плоского сопоставления, взаимодействующих с 8 этапами агрегирования. Jet выполнит это в общей сложности с 8 потоками (при условии, что у вас есть 8 availableProcessors), поэтому на уровне ОС почти не будет выполняться планирование потоков. Данные будут перемещаться между потоками большими порциями: flatMap ставит в очередь 1024 за раз, а затем каждый агрегатор извлекает все предназначенные для него элементы. Обмен очередями SPSC происходит без какого-либо вмешательства со стороны других потоков: каждый агрегирующий процессор имеет 8 входных очередей, по одной выделенной для каждого плоского сопоставления.

Во Flink каждый этап запускает еще 8 потоков, и я также заметил, что приемник имеет параллелизм 8, так что это 24 потока и еще один для источника. ОС должна будет запланировать их на 8 физических ядрах. Коммуникация будет происходить через очереди с несколькими производителями и одним потребителем (MPSC), что означает, что все потоки плоского сопоставления должны координироваться, чтобы только один поток в каждый момент времени ставил элемент в очередь любому заданному агрегатору, а конфликты приводили к горячим петлям CAS в все нити.

Чтобы подтвердить это подозрение, попробуйте собрать данные профилирования. Если приведенная выше история верна, вы должны увидеть, что Flink тратит много процессорного времени на постановку данных в очередь.

person Marko Topolnik    schedule 20.04.2021