En uno de los proyectos, era necesario transferir el proceso de importación de datos de sistemas de terceros a una arquitectura de microservicios. Apache NiFi fue elegido como herramienta. Como primer experimento, se seleccionó la importación del registro del Servicio Federal de Impuestos.
Los datos USRLE se publican como archivos XML empaquetados en archivos ZIP. Los archivos se presentan diariamente en un recurso https://ftp.egrul.nalog.ru/en un directorio separado para la fecha correspondiente. Para el acceso, la clave es # PKCS12.
La tarea que debe resolverse con NiFi es descargar archivos del Servicio de Impuestos Federales y preparar los datos descargados para importarlos a nuestros servicios. Este artículo describe cómo implementar descargas de archivos.
Fuente de datos
La obtención de datos de la USRLE se lleva a cabo como parte del servicio del Servicio de Impuestos Federales "Integración y acceso a las bases de datos de la USRLE y la USRIP". Aquí se presenta una descripción del modelo de interacción .
Este es el recurso del Servicio de Impuestos Federales, desde el cual desea descargar archivos.
Los directorios con el sufijo COMPLETO son la descarga del registro completo al comienzo del año correspondiente. Los catálogos restantes son actualizaciones diarias en el registro. Estamos interesados en descargar actualizaciones diarias.
Configuración de transmisión en Apache NiFi
La tarea de la transmisión es recopilar una lista de archivos de catálogo con la carga de ayer, obtener estos archivos y descomprimirlos.
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')})}
Aquellos. la fecha actual se toma y se convierte en una representación numérica de la fecha. Se le restan 86,400,000 milisegundos. El resultado se convierte en una representación de cadena de la fecha en el formato dd.MM.aaaa. La fecha resultante se agrega a la parte permanente del enlace.En la salida, obtenemos un FlowFile de la siguiente forma:
Capturas de pantalla FlowFile Recuperando Contenido del Directorio
El procesador InvokeHTTP se utiliza para obtener el contenido del directorio . Realiza una solicitud GET utilizando el enlace al directorio con la carga de ayer. En respuesta, el procesador recibe el código HTML del catálogo y agrega este código HTML al FlowFile como contenido.
Capturas de pantalla del procesador 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. .
En la cadena de certificados, debe seleccionar el certificado del Servicio de Impuestos DPC de Rusia y exportarlo en formato .CER con codificación DER . A continuación, debe importar el certificado del archivo recibido en el repositorio de cacerts utilizando la utilidad keytool . Por ejemplo, así:
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}
La contraseña predeterminada para cacerts es 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.
Capturas de pantalla del procesador UnpackContent En la salida, el procesador crea un FlowFile para cada archivo XML desempaquetado del archivo ZIP.
Más lejos...
Además, cada XML debe convertirse a JSON y desglosarse por organización, porque cada XML contiene de 1 a 1000 declaraciones de registro. Y desde JSON en el futuro será posible cargar datos en almacenamiento SQL o NoSQL.
La conversión de XML a JSON y AVROSchema se encuentra en el siguiente artículo.