Em um dos projetos, havia a necessidade de transferir o processo de importação de dados de sistemas de terceiros para uma arquitetura de microsserviço. O Apache NiFi foi escolhido como ferramenta. Como primeira experiência experimental, a importação do registro do Federal Tax Service foi selecionada.
Os dados USRLE são publicados como arquivos XML empacotados em arquivos ZIP. Os arquivos são organizados diariamente em um recurso https://ftp.egrul.nalog.ru/em um diretório separado para a data correspondente. Para acesso, a chave é # PKCS12.
A tarefa que precisa ser resolvida com o NiFi é fazer o download de arquivos do Serviço Fiscal Federal e preparar os dados baixados para importação em nossos serviços. Este artigo descreve como implementar downloads de arquivos.
Fonte de dados
A obtenção de dados do USRLE é realizada como parte dos bancos de dados do Serviço Federal de Tributação "Integração e Acesso ao Cadastro Estadual Unificado de Pessoas Jurídicas e USRIP". Uma descrição do modelo de interação é apresentada aqui .
Esse é o recurso do Federal Tax Service, do qual você deseja baixar arquivos.
Os diretórios com o sufixo CHEIO são o descarregamento do registro completo no início do ano correspondente. Os catálogos restantes são atualizações diárias no registro. Estamos interessados em baixar atualizações diárias.
Configuração de fluxo no Apache NiFi
A tarefa do fluxo é coletar uma lista de arquivos de catálogo com o upload de ontem, obter esses arquivos e descompactá-los.
NiFi :
- FlowFile,
- HTML
- HTML ,
FlowFile
FlowFile GenerateFlowFile.
24 FlowFile fnsEgrulURL. https://ftp.egrul.nalog.ru/?dir=EGRUL/14.04.2020. NiFi Expression Language:
${literal('https://ftp.egrul.nalog.ru/?dir=EGRUL/'):append(${now():toNumber():minus(86400000):format('dd.MM.yyyy')})}
Essa. a data atual é obtida e convertida em uma representação numérica da data. 86.400.000 milissegundos são subtraídos dele. O resultado é convertido em uma representação de sequência da data no formato dd.MM.aaaa. A data resultante é adicionada à parte permanente do link.A saída é um FlowFile do seguinte formato:
Recuperando conteúdo do diretório
O processador InvokeHTTP é usado para obter o conteúdo do diretório . Ele executa uma solicitação GET usando o link para o diretório com o upload de ontem. Em resposta, o processador recebe o código HTML do catálogo e adiciona esse código HTML ao FlowFile como conteúdo.
Imagens de tela do processador InvokeHTTP :
HTTPMethod — GET;
Remote URL — , URL . ${fnsEgrulURL} — FlowFile fnsEgrulURL;
SSL Context Service — SSLContextService , .. HTTPS. #PKCS12 .
FlowFile , , — HTML- .
SSLContextService
SSLContexService #PKCS12 , , .
cacerts JDK. . https://fns.egrul.nalog.ru , #PKCS12. .
Na cadeia de certificados, você deve selecionar o certificado de Serviço de imposto DPC russo e exportá-lo no formato .CER na codificação DER . Em seguida, você precisa importar o certificado do arquivo recebido para o repositório cacerts usando o utilitário keytool . Por exemplo, assim:
C:\Program Files\Java\jdk1.8.0_121\bin> keytool -importcert -keystore "C:\Program Files\Java\jdk1.8.0_121\jre\lib\security\cacerts" -file { .CER}
A senha padrão para cacerts é changeit .cacerts , NiFi . , Persistent Volume. SSLContextService. PKSC12, cacerts — JKS.
GetHTMLElement, HTML- FlowFile-. ZIP-.
<div id="page-content" class="container">
<div id="directory-list-header">
<div class="row">
<div class="col-md-7 col-sm-6 col-xs-10"></div>
<div class="col-md-2 col-sm-2 col-xs-2 text-right"></div>
<div class="col-md-3 col-sm-4 hidden-xs text-right"> </div>
</div>
</div>
<ul id="directory-listing" class="nav nav-pills nav-stacked">
<li data-name=".." data-href="https://ftp.egrul.nalog.ru/?dir=EGRUL">
<a href="https://ftp.egrul.nalog.ru/?dir=EGRUL" class="clearfix" data-name="..">
<div class="row">
<span class="file-name col-md-7 col-sm-6 col-xs-9">
<i class="fa fa-level-up fa-fw"></i>
.. </span>
<span class="file-size col-md-2 col-sm-2 col-xs-3 text-right">
- </span>
<span class="file-modified col-md-3 col-sm-4 hidden-xs text-right">
2020-04-05 22:00:00 </span>
</div>
</a>
</li>
<li data-name="EGRUL_2020-04-05_1.zip" data-href="EGRUL/05.04.2020/EGRUL_2020-04-05_1.zip">
<a href="EGRUL/05.04.2020/EGRUL_2020-04-05_1.zip" class="clearfix" data-name="EGRUL_2020-04-05_1.zip">
<div class="row">
<span class="file-name col-md-7 col-sm-6 col-xs-9">
<i class="fa fa-file-archive-o fa-fw"></i>
EGRUL_2020-04-05_1.zip </span>
<span class="file-size col-md-2 col-sm-2 col-xs-3 text-right">
528.78KB </span>
<span class="file-modified col-md-3 col-sm-4 hidden-xs text-right">
2020-04-05 22:00:24 </span>
</div>
</a>
<a href="javascript:void(0)" class="file-info-button">
<i class="fa fa-info-circle"></i>
</a>
</li>
</ul>
</div>
:
URL — URL HTML-;
CSS Selector — . li[data-name^=EGRUL] — li, data-name, EGRUL;
Output Type — Attribute — HTML-;
Destination — flowfile-attribute — FlowFile- ( HTMLElement);
Attribute Name — , . abs:${literal('data-href')} — URL (abs:) + data-href , CSS-.
, CSS- FlowFile ZIP- HTMLElement.
InvokeHTTP , HTML- . URL HTMLElement, ZIP-. SSLContextService .
ZIP- FlowFile .
UnpackContent. — ZIP.
Imagens de processador UnpackContent Na saída, o processador cria um FlowFile para cada arquivo XML descompactado do arquivo ZIP.
Mais longe...
Além disso, cada XML precisa ser convertido em JSON e dividido por organização, porque cada XML contém de 1 a 1000 instruções de registro. E a partir do JSON no futuro, será possível carregar dados no armazenamento SQL ou NoSQL.
A conversão de XML para JSON e AVROSchema está no próximo artigo.