Configurar el conector en NiFi

En esta lección, configurará el conector que ha creado en el paso anterior utilizando NiFi.

Before you begin

Descargue y extraiga el archivo SampleVersionedConnectors9112.zip en un directorio temporal en su entorno de desarrollo. Este archivo contiene un archivo JSON de ejemplo para los conectores versionados.

About this task

Utilice el enlace siguiente para NiFi: http://hostname/IP:30600/nifi/

Note:

Esta lección utiliza el nombre de conector custom.cas. Asegúrese de sustituir custom.cas por el nombre del conector que ha creado en el paso anterior.
Si está en la página de inicio de NiFi donde puede ver todos los procesadores, es posible que tenga que desplazarse a la derecha de la canalización y acercar el zoom para encontrar los procesadores necesarios.

Procedure

Para configurar el conector en NiFi, realice los pasos siguientes:

Desplácese al lado derecho del lienzo NiFi. El conector custom.cas que ha creado en el paso anterior se encuentra aquí.
Actualice las propiedades del procesador custom-_Template-Schema-CAS.
1. Vaya al nuevo canal custom.cas y haga doble clic en custom-_Template-Schema-CAS.
2. Haga doble clic en el grupo de procesos Actualización de esquema.
3. Busque el grupo de procesos Configurar esquema de índices de Elasticsearch y haga doble clic en él.
4. Busque y haga clic con el botón derecho en Specify Schema to Update y detenga el procesador.
5. Haga doble clic en el procesador para modificar la configuración.
6. Vaya a la pestaña Propiedades y actualice la propiedad param.schema con el valor product.
7. Pulse Aplicar.
8. Inicie el procesador pulsando con el botón derecho del ratón y seleccionando Iniciar.

Actualice el procesador Populate Index Schema.

Detener el procesador.
Haga doble clic en el procesador para modificar la configuración.

Actualice la propiedad Valor de sustitución con el valor siguiente:


{
   "properties":{
      "x_custom":{
         "properties":{
            "x_profitMargin":{
               "type":"float",
               "coerce":"true",
               "doc_values":"true",
               "ignore_malformed":"false",
               "index":"true",
               "store":"true"
            }
         }
      }
   }
}

Iniciar procesador.

Abra Flujo de NiFi > custom.cas- _Template-DatabaseETL-CAS > Canal de conector personalizado > Procesador del canal de conector personalizado
Arrastre el botón Procesador a un espacio en blanco en el lienzo y, a continuación, escriba Personalizado. Seleccione CustomFieldDocumentProcessor y pulse el botón ADD.
Añada CustomFieldDocumentProcessor, tal y como se muestra en la siguiente captura de pantalla de ejemplo.
Note: Debe detener los procesadores correspondientes para cambiar las relaciones de flujo.
Seleccione la pestaña de propiedades CustomFieldDocumentProcessor y rellene los valores de las propiedades siguientes:
- Database Connection Pooling Service: Database Connection Pool
- SQL Pre-Query: ${ingest.database.schema.sql}
- SQL select query: ${ingest.database.sql}

Abra Definir procesador SQL personalizado. Actualice el procesadorDefine custom SQL and index name. Actualice ingest.database.sql con el SQL siguiente:

SELECT
    OP.CATENTRY_ID,
    CASE
        WHEN OP.PRICE <> 0
        THEN (OP.PRICE - CP.PRICE) / OP.PRICE * 100
        ELSE NULL
    END AS PROFIT_MARGIN
FROM
    (SELECT O.CATENTRY_ID CATENTRY_ID, P.PRICE PRICE
      FROM OFFER O
         INNER JOIN OFFERPRICE P ON (O.OFFER_ID = P.OFFER_ID AND P.CURRENCY = 'USD')
         INNER JOIN TRADEPOSCN ON (O.TRADEPOSCN_ID = TRADEPOSCN.TRADEPOSCN_ID AND TRADEPOSCN.NAME = 'Extended Sites Catalog Asset Store')
      WHERE (O.STARTDATE IS NULL OR CURRENT_TIMESTAMP > O.STARTDATE)
       AND (O.ENDDATE IS NULL OR O.ENDDATE > CURRENT_TIMESTAMP)
       AND O.PUBLISHED = 1
    ) OP,
    (SELECT OFFER.CATENTRY_ID CATENTRY_ID, OFFERPRICE.PRICE PRICE
         FROM OFFER
         INNER JOIN OFFERPRICE ON (OFFER.OFFER_ID = OFFERPRICE.OFFER_ID AND OFFERPRICE.CURRENCY = 'USD')
         INNER JOIN TRADEPOSCN ON (OFFER.TRADEPOSCN_ID = TRADEPOSCN.TRADEPOSCN_ID AND TRADEPOSCN.NAME = 'My Company Cost Price List' )
    ) CP
WHERE OP.CATENTRY_ID = CP.CATENTRY_ID AND OP.CATENTRY_ID IN (${data.catentryId})

Anote la cláusula OP.CATENTRY_ID IN (${data.catentryId}) en el SQL. Cada flujo procesará un lote de entradas de catálogo.

El resultado de este SQL contiene dos columnas, CATENTRY_ID y PROFIT_MARGIN.

Note: Asegúrese de que los datos de precio de coste estén disponibles en su base de datos. Este conector no verifica si el SQL devuelve algún resultado. Si los datos del precio de coste no están disponibles en la base de datos, los datos devueltos para el SQL están vacíos y generan errores.

Cree una nueva versión para custom-cas_Template-Schema-CAS.
Haga clic con el botón derecho del ratón en el grupo de procesos custom-_Template-Schema-CAS.
Pulse Versión > Detener control de versión > Desconectar.
Haga clic con el botón derecho del ratón en el grupo de procesos custom-_Template-Schema-CAS.
Pulse Versión > Iniciar control de versión.
Agregue el Flow Name requerido y pulse Guardar.
En este tutorial, el Flow Name = ProfitMarginSchemaUpdateConnector.
Cree una nueva versión para custom.cas - _Template-DatabaseETL-CAS.

Haga clic con el botón derecho del ratón en el grupo de procesos custom -_Template-Groovy-DatabaseETL-CAS.

Pulse Versión > Detener control de versión > Desconectar.

Haga clic con el botón derecho del ratón en el grupo de procesos custom -_Template-Groovy-DatabaseETL-CAS.

Haga clic en Versión > Detener control de versión.

Agregue el nombre de flujo requerido y guarde.

En este tutorial, el Flow Name = ProfitMarginDatabaseConnectorPipe.
Copie el grupo de procesos custom.cas - _Template-Schema-CAS y péguelo en un espacio en blanco junto a auth.reindex.cas - Esquema de espacios de trabajo.

Inserte este nuevo grupo de procesos personalizado entre el esquema de espacios de trabajo y el siguiente grupo de procesos, tal y como se indica a continuación:

La secuencia del grupo de procesos puede cambiar. El grupo de procesos personalizado debe insertarse tras el esquema de espacios de trabajo.

Haga clic con el botón derecho del ratón en el grupo de procesos custom.cas - _Template-Schema-CAS y seleccione Iniciar para iniciar el grupo de procesos.
Siga unos pasos similares para añadir el grupo de procesos _Template-DatabaseETL-CAS antes del grupo de procesos Enviar producto a Elasticsearch.
Haga clic con el botón derecho del ratón en el grupo de procesos _Template-DatabaseETL-CAS y seleccione Iniciar para iniciar el grupo de procesos.
Note: Si ve un componente no válido en el grupo de procesos _custom.cas - _Template-DatabaseETL-CAS, siga los pasos que se indican a continuación:
Haga clic con el botón derecho del ratón en el grupo de procesos custom.cas - _Template-DatabaseETL-CAS y seleccione Configurar.

Vaya a Servicios del controlador

Pulse Habilitar para activafr los servicios de Agrupación de conexiones de la base de datos.
- Pulse el botón Habilitar y, para salir de la ventana emergente Habilitar servicios del controlador, pulse Cerrar.
- Haga clic con el botón derecho del ratón en el grupo de procesos custom.cas - _Template-DatabaseETL-CAS y seleccione Iniciar.
Extraiga el conector versionado.
1. Ejecute los mandatos siguientes:
  Note: El nombre del contenedor de Docker que se usa en este ejemplo es commerce_registry_1. Sustitúyalo por el nombre real del contenedor de Docker.
```
docker exec -it commerce_registry_1 bash

/opt/nifi-registry/scripts/export_flow.sh ProfitMarginSchemaUpdateConnector>/opt/nifi-registry/flows/ProfitMarginSchemaUpdateConnector.json

/opt/nifi-registry/scripts/export_flow.sh ProfitMarginDatabaseConnectorPipe>/opt/nifi-registry/flows/ProfitMarginDatabaseConnectorPipe.json

Exit

docker cp commerce_registry_1:/opt/nifi-registry/flows/ProfitMarginDatabaseConnectorPipe.json ./ProfitMarginDatabaseConnectorPipe.json

docker cp commerce_registry_1:/opt/nifi-registry/flows/ProfitMarginSchemaUpdateConnector.json ./ProfitMarginSchemaUpdateConnector.json
```
  Como alternativa, puede extraer el conducto de conector a través de la interfaz del registro de NiFi.
  1. Inicie la interfaz del registro de NiFi y localice el nombre de flujo que desea exportar, por ejemplo ProfitMarginSchemaUpdateConnector o ProfitMarginDatabaseConnectorPipe.
  2. Pulse el botón Acción y seleccione Exportar versión.
  3. Elija la versión más reciente y haga clic en Exportar.
2. Copie ProfitMarginSchemaUpdateConnector.json y ProfitMarginDatabaseConnectorPipe.json en el entorno de ejecución.