Время работы Apache Beam Wall продолжает увеличиваться
У меня есть конвейер Beam, который читает из темы pubsub, выполняет небольшие преобразования, а затем записывает события в некоторые таблицы BigQuery.
Преобразования просты в обработке, возможно, удаление поля или что-то еще, но, как вы можете видеть на изображении ниже, время стены для некоторых шагов очень велико. Что на самом деле может вызвать это?
Каждый элемент на самом деле является кортежем формы ((str, str, str), {**dict with data}). По этому ключу мы фактически пытаемся сделать наивную дедупликацию, взяв последнее событие по этому ключу.
В основном все, что я добавляю после этого Get latest element per key, работает медленно, и пометка также медленная, даже если она просто добавляет тег к элементу.
Ответы 1
Я предполагаю, что под «медленным» вы имеете в виду, сколько элементов он обрабатывает в секунду?
Здесь происходят две вещи. Во-первых, я предполагаю, что Get latest element per key содержит своего рода GroupByKey. Это включает в себя глобальное перемешивание, при котором все элементы отправляются по сети другим элементам, чтобы гарантировать, что все элементы с заданным ключом сгруппированы в одном рабочем потоке. Этот IO может быть дорогим, по крайней мере, с точки зрения времени стены.
Во-вторых, этапы, которые не нуждаются в обмене данными между работниками, «объединяются», что связывает их пропускную способность. Например. если за DoFnA следует DoFnB, за которым следует DoFnC, обработка продолжается, пропуская первый элемент через DoFnA, затем передавая эти выходные данные в DoFnB и затем DoFnC, прежде чем передать второй элемент в DoFnA. Это означает, что если одна из Fns (или чтение, или запись) имеет ограниченную пропускную способность, то все они будут ограничены.
На самом деле я не уверен, что есть место, где можно увидеть, какие этапы слились, но обычно это все между двумя GBK. Обычно вам не о чем беспокоиться.
Обновление: если щелкнуть шаг в пользовательском интерфейсе потока данных, в нижней части правой панели StepInfo появится раздел «Оптимизированные этапы». Шаги, которые объединены вместе, будут находиться на одном и том же оптимизированном этапе (например, F__). Обратите внимание, что некоторые шаги могут охватывать оптимизированные этапы.
Попался. А как насчет времени Стены? Должен ли я беспокоиться об этом? Я имею в виду, что задание выполняется в течение 48 часов, и у него есть несколько сумасшедших временных интервалов, например, этап Reshuffle имеет 50 дней, а значение тега - 13 дней... Какие-то сумасшедшие цифры, и я не уверен, что мне следует беспокоиться об этом :)
Walltime может быть полезным сигналом, но особенно если есть IO, я бы не стал об этом беспокоиться. (Потоковые конвейеры очень многопоточные, и это сумма времени прохождения каждого потока.)
Спасибо за объяснение. Да, эта группировка выполняется
combiners.Latest.PerKey(), и, по моим наблюдениям, с точки зрения количества элементов в секунду все в порядке. Основным узким местом является то, что следует за этим шагом. Сейчас та самая перестановка, до этого была пометка. У меня только один вопрос: - Как я узнаю, что какие-то шаги слиты воедино? Не знаю, поможет ли это, ошибка здесь - идентификатор работы2020-12-09_08_32_30-16230074399558339654:)