Apache NIFI — Criando Custom Processors + Debug Remoto

Apesar de o Apache NIFI ser uma ferramenta bastante completa, que possui processors de inúmeros tipos e que atendem à praticamente todas as necessidades básicas no desenvolvimento de fluxos de dados, talvez você se depare com cenários complexos onde não haja um processor capaz de resolver o seu problema.

Isto acontece, principalmente, quando se implementa muitas regras de negócio ao fluxo ou quando uma engine única/customizada precisa ser desenvolvida.

Pois bem, o NIFI também permite aos desenvolvedores criar seus próprios processors da maneira como eles quiserem, seguindo apenas algumas estruturas básicas de compatibilidade. Neste artigo, será mostrado um passo a passo de como criá-los e também como utilizar o debug remoto para realizar seus testes em runtime.

Criando seu primeiro Custom Processor

Antes de iniciar, certifique-se de ter instalada uma versão recente do JAVA JDK e MAVEN.

1. No diretório onde será criado o processor, executar o seguinte comando:

$ mvn archetype:generate

Em seguida:

• Digite “nifi” para filtrar os arquetipos pertinentes ao nifi;
• Digite o número equivalente à opção: org.apache.nifi:nifi-processor-bundle-archetype;
• Escolha a versão do NIFI para seu processor;
• Digite o groupId para seu processor (Exemplo: “intelipost.nifi”);
• Digite o artifactId para seu processor (Exemplo: “nifi-ipprocessor-bundle”);
• Digite a versão do seu processor (Exemplo: 1.0);
• Digite o artifactBaseName para seu processor (Exemplo: “ipprocessor”);
• Digite o nome do package para seu processor. É recomendado aceitar a opção padrão que o archetype te oferece. (Exemplo: “intelipost.nifi.processors.ipprocessor”);

Confirme se as opções acima estão corretas e digite “Y” para criar o arquetipo do processor.

2. Entre no diretório onde a estrutura foi criada, e utilize o maven para carregar as dependências:

$ cd nifi-ipprocessor-bundle

$ mvn install

Por padrão, o arquetipo será criado com um processor exemplo chamado MyProcessor e com a classe de teste MyProcessorTest. Você pode removê-los e criar as classes para seu próprio processor, seguindo a estrutura esperada pelo NIFI (falarei sobre ela mais adiante neste artigo);

3. Para que seu processor possa ser listado na lista de processors do NIFI, você deve alterar o arquivo “META-INF” e incluir a referência para seu processor:

nifi-ipprocessor-bundle/nifi-ipprocessor-processors/src/main/resources/META-INF.services/org.apache.nifi.processor.Processor

4. Para gerar o .NAR do seu processor (arquivo de pacote entendido pelo NIFI), execute o seguinte comando na pasta onde seu arquetipo foi criado:

$ mvn clean install

Feito isso, basta copiar o arquivo .NAR gerado, que fica localizado como indicado abaixo:

nifi-ipprocessor-bundle/nifi-ipprocessor-nar/target/nifi-ipprocessor-nar-1.0.nar

Dentro da pasta onde o NIFI está instalado, crie uma pasta chamada custom_lib e cole seu arquivo .NAR (nesta pasta ficarão guardados os custom processors criados. É altamente recomendado que os custom processor sejam depositados nesta pasta, e não diretamente na pasta lib).

Para que o NIFI reconheça os custom processors, dentro do arquivo conf/nifi.properties, adicionar a seguinte propery abaixo da linha nifi.nar.library.directory=./lib:

nifi.nar.library.directory.custom=./custom_lib

Agora basta reiniciar o NIFI e procurar por seu processor na lista de processors :)

Estrutura de um Custom Processor

Após criar um custom processor, você verá uma estrutura como a seguinte:

Dentro da classe MyProcessor (ou o nome dado por você) é onde irá ocorrer toda a lógica de programação do seu custom processor. Esta classe, bem como a de qualquer processor, segue uma estrutura padrão conhecido pelo NIFI, e implica nos principais pontos a seguir:

Properties

As properties de um processor são aquelas que aparecem na aba de configurações de um processor na interface do NIFI, onde o usuário consegue setar valores para cada property e com isso alterar o comportamento do processor.

Todas as particularidas a respeito das properties, como criar validadores, valores default, etc, podem ser encontrados na documentação oficial do NIFI:

https://nifi.apache.org/developer-guide.html

Relationships

Os relationships de um processor indicam os caminhos possíveis/alternativos que um flowfile pode seguir após ser processado pelo seu processor. A maioria dos processors utiliza dois relationships: SUCESS e FAILURE. Mas isso não impede que outros tipos de relationships sejam criados. Tudo depende da necessidade do negócio.

OnTrigger

É aqui que a magia acontece.

Quando um flowfile chega ao seu processor através do fluxo de dados, o que irá acontecer com ele se define aqui neste método “OnTrigger”. Ou seja, ao receber um flowfile, este método é invocado pelo framework do NIFI e executa a lógica presente no mesmo.

Este método recebe um ProcessContext, de um onde você consegue recuperar algumas informações como as configurações das Properties deste processor; e um ProcessSession, de onde você consegue recuperar o flowfile recebido e também fazer manipulações do fluxo deste flowfile, incluindo novos atributos, movendo para relationships específicos, entre outras operações.

Mais informações sobre custom processors podem ser encontradas no link a seguir:

https://www.nifi.rocks/developing-a-custom-apache-nifi-processor-json/

Como debugar um Custom Processor em runtime no NIFI

Para habilitar o debug remoto do NIFI, é necessário editar a configuração bootstrap localizada aqui em conf/bootstrap.conf

Descomente a seguinte linha (onde 8085 é a porta exemplo que será utilizada para o debug remoto):

java.arg.debug=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8085

Certifique-se de que o NIFI esteja rodando antes de ir para os próximos passos.

OBS: A IDE utilizada neste tutorial para fazer o debug remoto foi o IntelliJ, mas o princípio é mesmo para as demais IDEs, mudando apenas os locais onde se configura o debug.

1. Importe o código fonte do seu processor na sua IDE;
2. Crie uma configuração para Run/Debug mirando na sua instância do NIFI:

• Menu Run > Edit Configurations;
• Clique no botão “+” e selecione “Remote” no drop down;
• Crie uma nova configuração chamada “Debug Local NIFI”;
• Altere o campo “Port” para a mesma porta de debug remoto configurada no arquivo bootstrap.conf (no caso exemplo, 8085);
• Salve a configuração.

Caso tudo tenha sido configurado corretamente, ao rodar o seu projeto em debug a seguinte mensagem deverá aparecer em seu console:

Connected to the target VM, address: 'localhost:8085', transport: 'socket'

Agora basta colocar os breakpoints nos locais desejados e debugar.