Разница между shuffle() и rebalance() в Apache Flink
Я работаю над последним проектом моего бакалавра, который посвящен сравнению Apache Spark Streaming и Apache Flink (только потоковая передача), и я только что пришел к разделу "Физическое разбиение" в документации Flink. Дело в том, что в этой документации не очень хорошо объясняется, как работают эти два преобразования. Прямо из документации:
shuffle(): Разделение элементов случайным образом в соответствии с равномерным распределением.
rebalance(): Круглые элементы разделов, создающие равную нагрузку на раздел. Полезно для оптимизации производительности при наличии перекоса данных.
Источник: https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/datastream_api.html.
Оба автоматически выполняются, так что я понимаю, что они оба перераспределяют одинаково (shuffle() > равномерное распределение и rebalance() > Round-Robin) и случайным образом данные. Тогда я делаю вывод, что rebalance() лучше распределяет данные ("равная нагрузка на разделы"), поэтому задачи должны обрабатывать одинаковое количество данных, но shuffle() может создавать большие и меньшие разделы. Затем, в каких случаях вы можете использовать shuffle() чем rebalance() ?
Единственное, что приходит мне в голову, это то, что, вероятно, rebalance() требует некоторого времени обработки, поэтому в некоторых случаях может потребоваться больше времени для перебалансировки, чем время, которое он улучшит в будущих преобразованиях.
Я искал это, и никто не говорил об этом, только в списке рассылки Flink, но они не объясняют, как shuffle() работает.
Спасибо Sneftel, который помог мне улучшить мой вопрос и попросил у меня вещи, которые позволили бы мне пересмотреть то, что я хотел спросить; и Till Rohrmann, который довольно хорошо ответил на мой вопрос. : D
1 ответ
Как говорится в документации, shuffle будет случайным образом распределять данные, тогда как rebalance будет распространять данные в циклическом порядке. Последнее более эффективно, так как вам не нужно вычислять случайное число. Более того, в зависимости от случайности, вы можете получить какое-то неравномерное распределение.
С другой стороны, rebalance всегда начнет отправку первого элемента на первый канал. Таким образом, если у вас есть только несколько элементов (меньше элементов, чем подзадач), то только некоторые из подзадач получат элементы, потому что вы всегда начинаете отправлять первый элемент в первую подзадачу. В случае потоковой передачи это в конечном итоге не должно иметь значения, поскольку у вас обычно есть неограниченный входной поток.
Фактическая причина, по которой существуют оба метода, исторически обоснована. shuffle был представлен первым. Чтобы сделать пакет потоковым API более похожим, rebalance был тогда введен.
Это заявление Флинка вводит в заблуждение:
Полезно для оптимизации производительности при наличии перекоса данных.
Поскольку он используется для описания rebalance, но нет shuffle, это предполагает, что это отличительный фактор. Насколько я понимаю, если некоторые элементы обрабатываются медленно, а некоторые быстро, разделитель будет использовать следующий свободный канал для отправки элемента. Но это не так, сравните код для rebalance а также shuffle. Вrebalance просто добавляет на следующий канал, независимо от того, насколько он занят.
// rebalance
nextChannelToSendTo = (nextChannelToSendTo + 1) % numberOfChannels;
// shuffle
nextChannelToSendTo = random.nextInt(numberOfChannels);
Утверждение можно понимать и по-другому: "загрузка" означает не фактическое время обработки, а только количество элементов. Если исходное разбиение на разделы имеет перекос (значительно различающееся количество элементов в разделах), операция будет назначать элементы разделам равномерно. Однако в этом случае это относится к обеим операциям.
Мой вывод: shuffle а также rebalance сделай то же самое, но rebalanceделает это немного эффективнее. Но разница настолько мала, что вряд ли вы ее заметите,java.util.Random может сгенерировать 70 миллионов случайных чисел в одном потоке на моей машине.