Procesador personalizado de Apache Nifi procesando solo un registro de archivo.
He creado un procesador personalizado para procesar registros JSON de un archivo. El archivo de entrada tiene múltiples registros JSON separados por una nueva línea. Estoy utilizando los procesadores GetFile, CustomProcessor y PutFile. Sin embargo, solo obtengo un registro en el archivo de salida.
Utilicé el ejemplo de nifi disponible en GitHub en https://www.nifi.rocks/developing-a-custom-apache-nifi-processor-json/.
Mi nifi se ve así:
Me gustaría saber cómo puedo hacer que el procesador procese varios registros del archivo de entrada y escribir un número finito de registros en el archivo de salida.
davy.ai
Para procesar múltiples registros del archivo de entrada y escribir un número finito de registros en el archivo de salida, puedes hacer uso de los atributos de los flujo de datos de NiFi. En tu procesador personalizado, puedes establecer el número máximo de registros a escribir en el archivo de salida como un atributo. Luego, puedes utilizar un script o código para leer el archivo de entrada línea por línea, extraer los registros JSON y crear un flujo de datos para cada registro. Para cada flujo de datos, puedes agregar el número de registros procesados como un atributo.
Una vez que tengas el flujo de datos con JSON en cada registro, puedes utilizar un procesador divisor para dividir el flujo de datos en registros individuales. Luego, puedes utilizar otro procesador personalizado o un conjunto de procesadores para realizar cualquier transformación, filtrado o agregación requerida. Finalmente, los archivos de flujo de salida se pueden fusionar utilizando un procesador MergeContent y escribir en el archivo de salida utilizando el procesador PutFile.
Aquí tienes algunos pasos que puedes seguir para lograr esto:
Lee el archivo de entrada línea por línea, extrayendo cada registro JSON y creando un flujo de datos para cada registro. Para cada flujo de datos, agrega el número de registros procesados como un atributo.
Divide el flujo de datos en registros individuales utilizando el procesador SplitJson.
Utiliza tu procesador personalizado o un conjunto de procesadores para realizar cualquier transformación, filtrado o agregación requerida.
Fusiona los archivos de flujo de salida en un solo flujo de datos utilizando el procesador MergeContent.
Escribe el flujo de datos fusionado en el archivo de salida utilizando el procesador PutFile.
Con estos pasos, ahora deberías poder procesar múltiples registros del archivo de entrada y escribir un número finito de registros en el archivo de salida.