Olá a todos!

A tarefa consiste em — há um flow, apresentado na imagem acima, que precisa ser implantado em N servidores com Apache NiFi. O flow é um teste — gera um arquivo e o envia para outro instância do NiFi. A transferência de dados ocorre por meio do protocolo NiFi Site to Site.

NiFi Site to Site (S2S) — uma maneira segura e facilmente configurável de transferir dados entre instâncias do NiFi. Como funciona o S2S, veja em documentação e é importante não esquecer de configurar a instância do NiFi para permitir o S2S, veja aqui.

Nos casos em que se trata da transferência de dados usando S2S — uma instância é chamada de cliente, a outra de servidor. O cliente envia os dados, o servidor os recebe. Existem duas maneiras de configurar a transferência de dados entre eles:

1. Push. Da instância cliente, os dados são enviados usando Remote Process Group (RPG). Na instância do servidor, os dados são recebidos através de Input Port
2. Pull. O servidor recebe os dados usando RPG, o cliente envia usando Output port.

O flow a ser implantado é armazenado no Apache Registry.

Apache NiFi Registry — um subprojeto do Apache NiFi, que representa uma ferramenta para armazenar flows e gerenciar versões. Algo como um GIT. Informações sobre instalação, configuração e uso do registry podem ser encontradas em documentação oficial. O flow a ser armazenado é agrupado em uma process group e armazenado desta forma no registry. Voltaremos a isso mais adiante no artigo.

No início, quando N é um número pequeno, o flow é entregue e atualizado manualmente em um tempo aceitável.

Mas com o aumento de N, os problemas aumentam:

1. a atualização do flow leva mais tempo. É preciso acessar todos os servidores
2. ocorrem erros de atualização dos templates. Aqui foi atualizado, mas aqui foi esquecido
3. erros humanos ao realizar um grande número de operações semelhantes

Tudo isso nos leva à conclusão de que precisamos automatizar o processo. Eu experimentei as seguintes maneiras de resolver essa tarefa:

1. Usar MiNiFi em vez de NiFi
2. NiFi CLI
3. NiPyAPI

Uso do MiNiFi

Apache MiNiFy — subprojeto Apache NiFi. MiNiFy — um agente compacto que utiliza os mesmos processadores que o NiFi, permitindo criar os mesmos fluxos que no NiFi. A leveza do agente é alcançada em parte pela ausência de uma interface gráfica para a configuração do fluxo. A falta de uma interface gráfica no MiNiFy significa que é necessário resolver o problema de entrega do fluxo para o minifi. Como o MiNiFy é amplamente utilizado no IOT, há muitos componentes e o processo de entrega do fluxo para as instâncias finais do minifi precisa ser automatizado. Uma tarefa familiar, não é?

Mais um subprojeto que pode ajudar a resolver essa tarefa é o MiNiFi C2 Server. Este produto é destinado a ser o ponto central na arquitetura de implantação de configurações. Como configurar o ambiente é descrito em este artigo Habr e há informação suficiente para resolver a tarefa proposta. O MiNiFi, em conjunto com o servidor C2, atualiza automaticamente sua configuração. O único ponto negativo dessa abordagem é que é necessário criar templates no C2 Server, um simples commit no registry não é suficiente.

A opção descrita no artigo acima é funcional e não é difícil de implementar, mas é preciso lembrar o seguinte:

1. No minifi não estão todos os processadores do nifi
2. As versões dos processadores no Minifi estão atrasadas em relação às versões dos processadores no NiFi.

No momento da redação desta publicação, a última versão do NiFi é 1.9.2. A versão dos processadores da última versão do MiNiFi é 1.7.0. Processadores podem ser adicionados ao MiNiFi, mas devido à discrepância de versões entre os processadores NiFi e MiNiFi, isso pode não funcionar.

NiFi CLI

De acordo com a descrição da ferramenta no site oficial, é uma ferramenta para automatizar a interação entre NiFi e NiFi Registry na entrega de fluxos ou gerenciamento de processos. Para começar a usar, essa ferramenta precisa ser baixada daqui.

Iniciamos a utilidade

.\/bin\/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  \/  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Digite 'help' para ver uma lista de comandos disponíveis, use tab para autocompletar.

Para que possamos carregar o fluxo necessário do registry, precisamos saber os identificadores do bucket (bucket identifier) e do próprio fluxo (flow identifier). Esses dados podem ser obtidos pelo cli ou na interface web do NiFi registry. Na interface web, aparece assim:

Com o CLI, é feito assim:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

Iniciamos a importação do grupo de processos do registry:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

Um ponto importante — como host, para o qual estamos implantando o grupo de processos, pode ser especificada qualquer instância do nifi.

O grupo de processos foi adicionado com os processadores parados, é necessário iniciá-los.

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

Ótimo, os processadores foram iniciados. No entanto, de acordo com as condições da tarefa, precisamos que as instâncias do NiFi enviem dados para outras instâncias. Suponhamos que, para a transmissão de dados para o servidor, a forma escolhida tenha sido Push. Para organizar a transmissão de dados, precisamos habilitar a transmissão de dados no Remote Process Group (RPG) adicionado, que já está incluído em nosso fluxo.

Na documentação, na CLI e em outras fontes, não encontrei um meio de habilitar a transmissão de dados. Se você souber como fazer isso, por favor, escreva nos comentários.

Uma vez que temos o bash e estamos prontos para ir até o fim — vamos encontrar uma solução! Podemos usar a API do NiFi para resolver esse problema. Vamos usar o seguinte método, ID retirado dos exemplos acima (neste caso, é 7f522a13-016e-1000-e504-d5b15587f2f3). Descrição dos métodos da API do NiFi. aqui.

No body, é preciso enviar um JSON da seguinte forma:

{
    "revision": {
	    "clientId": "valor",
	    "version": 0,
	    "lastModifier": "valor"
	},
    "state": "valor",
    "disconnectedNodeAcknowledged": true
}

Parâmetros que precisam ser preenchidos para que "funcione":
state — status da transmissão de dados. Disponível TRANSMITTING para habilitar a transmissão de dados, STOPPED para desativá-la.
version — versão do processador.

a versão por padrão será 0 ao ser criada, mas esses parâmetros podem ser obtidos usando o método.

Para os amantes de scripts em bash, esse método pode parecer útil, mas eu acho complicado — scripts em bash não são o que eu mais gosto. O próximo método é mais interessante e prático, na minha opinião.

NiPyAPI

NiPyAPI — biblioteca para a linguagem Python para interagir com instâncias do NiFi. Página com a documentação contém as informações necessárias para trabalhar com a biblioteca. O início rápido está descrito em no qual estou trabalhando agora, é necessário acessar hosts atrás de NAT externamente. Usando protocolos com criptografia madura, eu não conseguia me livrar da sensação de que isso é como usar uma bazuca para matar um passarinho. Como o túnel é usado principalmente apenas para perfurar um buraco no NAT, o tráfego interno geralmente também é criptografado, mas ainda preferem HTTPS. no github.

Nosso script para a implantação da configuração — um programa em Python. Vamos para a codificação.
Configuramos os configs para o trabalho futuro. Precisaremos dos seguintes parâmetros:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #caminho até a instância nifi-api, onde estamos implantando o grupo de processos
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #caminho até a nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeautifulRegistry' #nome do registry, como será chamado na instância do nifi
nipyapi.config.bucket_name = 'BucketName' #nome do bucket, do qual estamos puxando o fluxo
nipyapi.config.flow_name = 'FlowName' #nome do fluxo, que estamos puxando

A seguir, vou inserir os nomes dos métodos desta biblioteca que estão descritos aqui.

Conectamos o registry à instância do nifi usando

nipyapi.versioning.create_registry_client

Neste passo, também é possível adicionar uma verificação para saber se o registry já foi adicionado à instância; para isso, pode-se usar o método

nipyapi.versioning.list_registry_clients

Encontramos o bucket para a busca subsequente do flow no bucket

nipyapi.versioning.get_registry_bucket

Com o bucket encontrado, procuramos o flow

nipyapi.versioning.get_flow_in_bucket

Agora, é importante entender se este process group já foi adicionado. O process group é posicionado por coordenadas e pode ocorrer a situação em que um componente se sobreponha a outro. Eu verifiquei, isso pode acontecer 🙂 Para obter todos os process groups adicionados, usamos o método

nipyapi.canvas.list_all_process_groups

e então podemos procurar, por exemplo, pelo nome.

Não vou descrever o processo de atualização do template; direi apenas que se, na nova versão do template, os processadores são adicionados, não há problemas com a presença de mensagens nas filas. Porém, se os processadores forem removidos, problemas podem surgir (nifi não permite remover um processador se houver mensagens acumuladas antes dele na fila). Se você estiver interessado em como resolvi esse problema, por favor me escreva, vamos discutir isso. Contatos no final do artigo. Vamos ao passo de adicionar o process group.

Durante a depuração do script, encontrei uma particularidade: a última versão do flow nem sempre é puxada, portanto, recomendo primeiro verificar essa versão:

nipyapi.versioning.get_latest_flow_ver

Fazendo o deploy do process group:

nipyapi.versioning.deploy_flow_version

Iniciamos os processadores:

nipyapi.canvas.schedule_process_group

No bloco sobre CLI, foi mencionado que a transmissão de dados não é ativada automaticamente no remote process group? Enfrentei esse problema também ao implementar o script. Naquele momento, não consegui iniciar a transmissão de dados usando a API e decidi escrever para o desenvolvedor da biblioteca NiPyAPI e pedir conselhos/ajuda. O desenvolvedor me respondeu, discutimos o problema e ele mencionou que precisava de tempo para “verificar algumas coisas”. E então, após alguns dias, recebi um e-mail com uma função em Python que resolveu meu problema de ativação!!! Naquele momento, a versão do NiPyAPI era 0.13.3 e, claro, nada disso estava presente. Mas na versão 0.14.0, que foi lançada recentemente, essa função já veio incorporada à biblioteca. Conheçam,

nipyapi.canvas.set_remote_process_group_transmission

Assim, com a biblioteca NiPyAPI, conectamos o registry, implementamos o flow e até lançamos os processadores e a transmissão de dados. A partir daí, podemos ajustar o código, adicionar diversas verificações, fazer logging e tudo mais. Mas isso já é uma história completamente diferente.

Dentre as opções de automação que examinei, a última me pareceu a mais viável. Em primeiro lugar, é código em Python, que permite a incorporação de código auxiliar e o aproveitamento de todas as vantagens da linguagem. Em segundo lugar, o projeto NiPyAPI está em desenvolvimento ativo e, em caso de problemas, é possível contatar o desenvolvedor. Em terceiro lugar, o NiPyAPI é uma ferramenta mais flexível para interagir com o NiFi na solução de problemas complexos. Por exemplo, na verificação de se as filas de mensagens estão vazias no flow e se é possível atualizar o grupo de processos.

Isso é tudo. Apresentei 3 abordagens para a automação da entrega de flows no NiFi, os obstáculos que um desenvolvedor pode encontrar e forneci o código funcional para a automação da entrega. Se você está tão interessado nesse tema quanto eu, entre em contato!

Fonte: habr.com