Apache Beam / Google dataflow - Обработка ошибок

Question

У меня есть конвейер с несколькими шагами (чуть выше 15). Я хочу сообщать об ошибках каждый раз, когда DoFn не удается. Я начал реализовывать это через TupleTags с кодом, таким как:

try {
 ... do stuff ...
 c.output(successTag, ...);
} catch (Exception e) {
 c.output(failureTag, new Failure(...));
}

Но так как мой конвейер содержит много шагов, это делает код определения конвейера довольно сложным для чтения / сопровождения.

Есть ли более глобальный способ добиться этого? Что-то вроде создания пользовательского исключения, которое обрабатывается глобально на уровне конвейера?

Answer 1

То, что вы делаете, - это правильный подход к отлову ошибок и их выдаче по-разному.Это понадобится вам на каждом этапе.Вы можете использовать шаблон Java, чтобы использовать его повторно, если вы предпочитаете.Создайте базовый класс для всех ваших ParDos и в processElement добавьте код обработки исключений.Затем реализуйте ваш processElement в отдельной функции (т.е. processElementImpl), которую вы вызываете в processElement.