Как обрабатывать один поток файла в нифига?

Question

NiFi версия 1.5 (кластер из 3 узлов).

у меня поток nifi в виде ListFTP -> FetchFTP (1) -> RouteOnAttribute -> .... -> UpdateAttribute -> FetchFTP ( 2) -> ExecuteGroovyScript -> PutFTP

все работает отлично. когда потоковый файл поступает в FetchFTP (2), я хочу обрабатывать один потоковый файл за раз с интервалом 30 сек c. независимо от 3-х узлов параллельной обработки потокового файла.

что я пробовал:

1. Установка RunSchedule на 30se c.

   ПРОБЛЕМА: это не сработало, потому что каждый файл потока первоначально обрабатывался разными узлами. когда дело доходит до FetchFTP, три потоковых файла обрабатываются одновременно. Также пробовал режим выполнения «Основной» из «Все узлы». Это не сработало, поскольку FetchFTP обрабатывает только потоковый файл, который первоначально был обработан первичным узлом, а все остальные потоковые файлы (обработанные другим узлом) остаются в очереди.

2. с использованием ControlRate перед процессором FetchFTP. три потоковых файла обрабатываются одновременно.
3. с использованием процессора Wait и Notify. три потоковых файла обрабатываются одновременно.

- это то, что я пропускаю, или любой другой лучший способ сделать это.

Answer 1

Я хочу обрабатывать один файл потока одновременно с интервалом 30 сек c. независимо от 3-х узлов параллельная обработка потокового файла.

В панели конфигурации каждого процессора есть возможность выполнять на основном узле только при работе в кластерном режиме. Я не рекомендую его по соображениям производительности, но если вы уверены, что он вам нужен здесь, он должен сделать все возможное, чтобы обеспечить «1 поток, 1 процесс только» в кластере.

Answer 2

Используйте простой процессор между ListFtp и FetchFtp. Например, UpdateAttribute (Barrier на изображении ниже)

Установите балансировку нагрузки в очереди между Barrier и FetchFtp для балансировки файлов между узлами.

И установите для Barrier schedule=30 sec и выполняйте только на основном узле .

В качестве альтернативы вы можете играть с противодавлением для очередей.