Идеальный способ отладки сложных потоков данных NiFi

Question

Из того, что я понял после использования NiFi для создания некоторых PoC для приема в БД, весь поток данных работает как поток потоковых файлов. И в любой конкретный момент контроль выполнения может выполняться на одном или нескольких процессорах одновременно.

Так что я действительно запутался в том, как отлаживать сложный поток данных для любых сбоев.

Мой рабочий процесс PoC выглядит следующим образом.

И когда мы перейдем к производственным сценариям использования, это может оказаться намного сложнее, чем это. Поэтому у меня есть несколько вопросов.

1. Как узнать состояние потока данных. Если, скажем, 4 из 10 разветвленных потоковых файлов завершились с ошибкой GenerateTableFetch из-за ошибки пула базы данных, как узнать, какие из них не удалось, и как быстро воспроизвести их, не переходя к происхождению данных и не делая один за другим.

2. Есть ли способ узнать, просто посмотрев на поток данных, какие потоки на каком процессоре выходят из строя.

У меня гораздо больше сомнений / путаницы в отладке потоков данных с помощью NiFi, и если кто-то может подсказать мне какой-нибудь документ или поделиться передовым опытом, это будет полезно.

Спасибо.

Answer 1

1- Как узнать состояние потока данных.Например, если в GenerateTableFetch произошла ошибка 4 из 10 разветвленных потоковых файлов из-за ошибки пула базы данных, как узнать, какие из них не удалось, и как быстро воспроизвести их, не переходя к управлению данными и не делая по одному.

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

Так, как Брайан упомянул, вы нехотите, чтобы они автоматически прекращались, если вам все равно.

2- Есть ли способ узнать, просто посмотрев на поток данных, какие потоковые файлы на каком процессоре выходят из строя.

Да - вы должны установить «Бюллетень»level ", чтобы отклонить уровень логов

Как управлять вашими потоками NiFi, которые терпят неудачу?

Что ж, вам нужно быть лучшими друзьями с BuletinBoard, смотрите здесь SiteToSiteStatusReportingTask или вы можете использовать InvokeHttp против собственного NiFI Rest Api с вызовом GET для http://nifi -сервера:port / nifi-api / flow / bulletin-board , и это ответит подробным json-объектом, который можно проанализировать, а затем вставить в PutSlack / PutEmail / PutSNS для любой ошибки.

Также идеально иметь Shared Process Group для обработки любых входящих файлов потока ошибок. Этот PG будет построен с правилами и маршрутами, применяемыми ко всей логике потока данных на вашем сервере NiFi.Очень важно иметь специфичные для PG атрибуты, которые будут переноситься со всеми вашими потоками и использоваться в ходе потока данных.

Например:

Группа процессов "Демо" имеет процессор с именем Установить атрибуты PG , который устанавливает атрибут PGName , атрибут PGType , атрибут FailEmailTitle ,так далее.Если мой поток завершится ошибкой в ​​какой-то момент, отношение сбоев направит мой неудавшийся поток на основе значения одного из атрибутов, установленных в Установить атрибуты PG процессор

Вот диаграмма моего текущегоустановка, где у меня есть все ошибки отправлены на тот же общий PG.

Другой вариант

Если вы считаете, что сообщение, сохраняющееся в течение 5 минут, является проблемой, вы можете использовать nifi-app.log , который может быть настроен для заполнения по правилам в вашем / opt / nifi / conf / logback.xml файле

  <logger name="org.apache.nifi" level="ERROR"/>
    <logger name="org.apache.nifi.processors" level="DEBUG"/>
    <logger name="org.apache.nifi.processors.standard.LogAttribute" level="ERROR"/>
    <logger name="org.apache.nifi.processors.standard.LogMessage" level="ERROR"/>
    <logger name="org.apache.nifi.controller.repository.StandardProcessSession" level="ERROR" />

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

Answer 2

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

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