Медленная пропускная способность после groupBy

Question

Я заметил, что в моих задачах пропускная способность (количество зарегистрированных записей/сек) значительно замедляется после шага «группа за». Когда этот шаг рабочего процесса выполняется, я вижу, что в некоторых случаях использование процессора составляет ~ 30%, в то время как некоторые, похоже, простаивают.

Является ли это проблемой потока данных, или я должен как-то инструктировать рабочий процесс, чтобы увеличить параллельность этого шага?

Спасибо, G

Answer 1

Его трудно знать наверняка, что происходит, не зная больше подробностей о том, что делает ваш трубопровод.

В общем пропускной способности (количество записей/сек) зависит от нескольких факторов, таких как

• Размер записей.
• Объем переработки осуществляется вашим Пардо

В общем виде GroupByKey строит большую запись, состоящую из ключа и все значения с помощью этого ключа; т.е. вход представляет собой набор КВ < K, V> а выход представляет собой набор KV < К, Iterable < V >>

В результате, в целом я бы ожидать, что записи, выводимый в GroupByKey являются намного больше, чем входные записи. Поскольку записи больше, они занимают больше времени, поэтому записи/сек будут уменьшаться.

Низкое использование ЦП не является неожиданным в выпуске Alpha для потока данных Alpha. В настоящее время Dataflow не полностью использует преимущества всех ядер VM для обработки работы. Для улучшения этого улучшится ряд улучшений.

Dataflow в настоящее время предоставляет две ручки для настройки количества параллельности с помощью флагов

--numWorkers=<integer> 
--workerMachineType=<Name of GCE VM Machine Type> 

--numWorkers позволяет увеличить или уменьшить количество рабочих, используемых для обработки данных параллельно. В общем, увеличение числа рабочих позволяет параллельно обрабатывать данные.

Использование --workerMachineType позволяет выбрать машину с более или менее процессором или оперативной памятью.

Если вы заметили, что ваш процессор VM недостаточно используется, вы можете выбрать машину с меньшим количеством процессоров (по умолчанию Dataflow использует 4 процессора на ВМ). Если вы уменьшаете количество процессоров на одну машину, но увеличиваете число пользователей, чтобы общее количество процессоров было примерно одинаковым, вы могли бы увеличить количество параллелизма, не увеличивая стоимость своей работы.

В настоящее время Dataflow предоставляет только эти очень грубые ручки для управления количеством параллелизма на глобальном уровне (в отличие от уровня каждой ступени). Это может измениться в будущем. Однако в целом наша цель - автоматически настроить количество параллелизма для вас, поэтому вам не нужно.

Answer 2

Низкая пропускная способность также может быть результатом «горячих клавиш» или очень часто встречающихся ключей. Это приведет к созданию нескольких чрезвычайно больших коллекций, которые обрабатываются одним ядром у одного работника.

Here - официальная документация Google относительно горячих клавиш и способы их устранения.По моему опыту, выборочное применение фактора разветвления с использованием Combine.PerKeyWithHotKeyFanout дало хорошие результаты.