Olá a todos!
A tarefa consiste em — há um flow, apresentado na imagem acima, que precisa ser implantado em N servidores com . 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 e é importante não esquecer de configurar a instância do NiFi para permitir o S2S, veja .
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:
- 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
- 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 . 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:
- a atualização do flow leva mais tempo. É preciso acessar todos os servidores
- ocorrem erros de atualização dos templates. Aqui foi atualizado, mas aqui foi esquecido
- 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:
- Usar MiNiFi em vez de NiFi
- NiFi CLI
- NiPyAPI
Uso do MiNiFi
— 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 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:
- No minifi não estão todos os processadores do nifi
- 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 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 .
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. .
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. contém as informações necessárias para trabalhar com a biblioteca. O início rápido está descrito em 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 .
Conectamos o registry à instância do nifi usando
nipyapi.versioning.create_registry_clientNeste 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_clientsEncontramos o bucket para a busca subsequente do flow no bucket
nipyapi.versioning.get_registry_bucketCom o bucket encontrado, procuramos o flow
nipyapi.versioning.get_flow_in_bucketAgora, é 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_groupse 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_verFazendo o deploy do process group:
nipyapi.versioning.deploy_flow_versionIniciamos os processadores:
nipyapi.canvas.schedule_process_groupNo 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_transmissionAssim, 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,
Fonte: habr.com
