¿Como mover datos de tablas con NiFi?

En este post, vamos a realizar el movimiento de datos entre dos tablas de una base de datos Oracle. Pueden estar en la misma instancia o en diferentes, pero la idea es mostrar como hacerlo con NiFi.
Nuestras tablas de ejemplo. Como puede observarse, la tabla V1, tiene un campo menos que la tabla V2. En nuestro ejemplo, vamos a hacerlo, cuando el campo acepta nulos, y cuando el campo, como en este caso, not acepta nulos.

 

 Datos de ejemplo en la tabla V1:

 

¿Como hacer para insertar si la tabla destino tiene mas o menos campos, pero estos aceptan Nulos?

Este caso es el más simple, donde simplemente configuramos el PutDatabaseRecord para que ignore si la tabla destino no tiene todas las columnas, con lo cual, el valor quedará en nulo.

El AVROReader en este caso es simple, dado que la definición del esquema de los datos viene en el mismo AVRO, pero si tuvieramos un caso donde queremos ahorrar en el footprint de los metadatos del AVRO, podemos usar Schema Registry. Este es un repositorio central de esquemas, con versionamiento y que nos ayuda a tener centralizado el esquema de los datos.

 

El resultado final del flujo será algo parecido a esto, donde tenemos un ExecuteSQL que toma los datos de la tabla fuente, y el resultado es una serie de flowfiles en formato AVRO. Estos AVRO los tomamos con el PutDatabaseRecord y los interpretamos con su esquema embebido en el mismo AVRO, y llevado como INSERT a la tabla destino. Como configuramos que ignore si no tiene todas los campos necesarios, entonces, insertará los que tiene y los demás los ignora. Si la tabla acepta NULLs, esto resultará en un INSERT exitoso.

¿Como hacer para insertar si la tabla destino tiene mas o menos campos, pero estos no aceptan Nulos?


En este caso, debemos ejecutar una variante que es la conversión del AVRO. Por ejemplo, en nuestro es un campo EMAIL que no acepta nulos, por lo que vamos a agregar este processor en el medio del ExecuteSQL y PutDatabaseRecord para transformar el esquema y el de salida sea el que use para insertar en la tabla destino.

Error que marca al intentar Insertar

Agregamos un procesador de tipo ConvertAVROSchema

 

  

Este procesador nos pide un esquema de entrada y otro de salida. También podemos agregar reglas dinámicas que nos permiten traducir desde el esquema fuente al destino por ejemplo, nombre de campos que cambian.

 

Nuestro esquema de salida quedaría marcado por una propiedad de "Default" para el campo EMAIL. En nuestro caso pusimos que tuviera el valor "NOT SET".

{
   "type":"record",
   "name":"NiFi_ExecuteSQL_Record",
   "namespace":"any.data",
   "fields":[
      {
         "name":"EMPLOYEE_ID",
         "type":[
            "null",
            {
               "type":"bytes",
               "logicalType":"decimal",
               "precision":6,
               "scale":0
            }
         ]
      },
      {
         "name":"FIRST_NAME",
         "type":[
            "null",
            "string"
         ]
      },
      {
         "name":"LAST_NAME",
         "type":[
            "null",
            "string"
         ]
      },
      {
         "name":"SALARY",
         "type":[
            "null",
            {
               "type":"bytes",
               "logicalType":"decimal",
               "precision":8,
               "scale":2
            }
         ]
      },
      {
         "name":"EMAIL",
         "type":[
            "string"
         ],
         "default":"NOT SET"
      }
   ]
}




Finalmente el flujo estaría de esta manera:



Y el dato de salida hacia la tabla tendría este formato:

Y la tabla finalmente con los datos se vería así: