Configurar el conector en NiFi

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

About this task

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

Note:

Esta lección utiliza el nombre de conector "custom". Asegúrese de sustituir "custom" 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:

Actualice las propiedades del procesador custom-_Template-Schema.
1. Vaya al nuevo canal "personalizado" y haga doble clic en custom-_Template-Schema.
2. Haga doble clic en el grupo de procesos Actualización de esquema.
3. Busque el grupo de procesos Configurar esquema de índice Elasticsearch.
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. Haga clic en Aplicar.
8. Inicie el procesador pulsando con el botón derecho del ratón y seleccionando Iniciar.
Actualice el procesador Populate Index Schema.
1. Detener el procesador.
2. Haga doble clic en el procesador para modificar la configuración.
3. Actualice la propiedad Valor de sustitución con el valor siguiente:
```
 { "properties":{ "custom":{ "properties":{ "x_profitMargin":{ "type":"float", "coerce":"true", "doc_values":"true", "ignore_malformed":"false", "index":"true", "store":"true" } } } } } 
```
4. Iniciar procesador.
Actualice el procesadorDefine custom SQL and index name.

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. Para verificar que los datos de precio de coste estén presentes en la base de datos, verifique el conector de indexación en auth.reindex > DatabasePriceStage1 > Buscar precios > Ejecutar SQL con Precheck.
1. Vaya a la nueva canalización personalizada y pulse dos veces custom -_Template -DatabaseETL.
2. Vaya a Personalizar canalización de conector > Ejecutar SQL.
3. Dentro de Ejecutar grupo de procesos SQL, busque el proceso Define custom SQL.
4. Pulse con el botón derecho Define custom SQL y detenga el procesador.
5. Haga doble clic en el procesador para modificar la configuración.
6. Actualice la propiedad Ingest.database.sql con el valor 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 LEFT OUTER JOIN (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 ON (CP.CATENTRY_ID = CP.CATENTRY_ID) INNER JOIN CATENTRY C ON C.CATENTRY_ID = CP.CATENTRY_ID WHERE OP.CATENTRY_ID = CP.CATENTRY_ID ${extCatentryAndSQL} 
```
7. Pulse Aplicar.
8. Inicie el procesador pulsando con el botón derecho y seleccionando Iniciar en el menú contextual.
Actualice el procesador Transform Document - Map Index Fields From Database.
1. Vaya a custom -_Template -DatabaseETL > Custom Connector Pipe > Map Index Field From Database.
2. Pulse con el botón derecho en el procesador y deténgalo.
3. Haga doble clic en el procesador para modificar su configuración. Se abre la ventana Configurar procesadores.
4. En la pestaña Propiedades de la ventana Configurar procesadores, actualice la propiedad Identificador de entradas con el valor ${param.storeId}-${param.langId }-${param.catalogId}-[id.catentry].
5. Añada una nueva propiedad llamada PROFIT_MARGIN y asígnele el valor custom.x_profitMargin.
  Su ventana Configurar procesadores será similar a la siguiente:
6. Haga clic en Aplicar.
7. Abra la ventana Detalles del procesador, elimine las propiedades FIELD1 y FIELD2 EN la pestaña Propiedades. Pulse Aceptar.
Crear una nueva versión para el custom-_Template-Schema
1. Pulse el botón derecho del ratón en el grupo de procesos, custom-_Template-Schema.
2. Pulse Versión > Detener control de versión > Desconectar.
3. Pulse el botón derecho del ratón en el grupo de procesos, custom-_Template-Schema.
4. Pulse Versión > Iniciar control de versión.
5. Agregue el Flow Name requerido y pulse Guardar.
  En este tutorial, el Flow Name = ProfitMarginSchemaUpdateConnector.
Cree una nueva versión para custom -_Template-DatabaseETL.
1. Pulse con el botón derecho el grupo de procesos custom -_Template-DatabaseETL.
2. Pulse Versión > Detener control de versión > Desconectar.
3. Pulse con el botón derecho el grupo de procesos custom -_Template-DatabaseETL.
4. Pulse Versión > Iniciar control de versión.
5. Agregue el Flow Name requerido y pulse Guardar.
  En este tutorial, el Flow Name = ProfitMarginDatabaseConnectorPipe.

Una al conector personalizado a su auth.postindex o live.postindex con los siguientes pasos:

Usando la interfaz de usuario de Swagger, vaya a Configuración del conector > POST: /upgrade API.
Haga clic en Probar.
Introduzca id como auth.postindex.

Para el cuerpo de la API, utilice el código siguiente.

{ "Licensed Materials": "Property of HCL Technologies Limited. (c) Copyright HCL Technologies Limited 1996, 2023.", "_name": "Post re-indexing connector descriptor for HCL Commerce Ingest service to be used with the Authoring environment", "_usage": { "// This is the connector descriptor of a post reindexing connector to be used with HCL Commerce Ingest service.": "", "// It is designed for performing a list of post indexing operations after the main part of a full re-indexing is completed.": "" }, "name": "auth.postindex", "environment": [ "auth", "data", "toolkit", "live" ], "description": "This is the connector for customization to perform a post re-indexing operation.", "pipes": [ { "// ": "------------- Preparation Stage -------------", "// Preparing database specific settings for this indexing pipeline": "", "name": "ReindexLink", "label": "ReindexLink" }, { "name": "ProfitMarginSchemaUpdateConnector", "label": "ReindexLink-ProfitMarginSchemaUpdateConnector" }, { "name": "ProfitMarginDatabaseConnectorPipe", "label": "ReindexLink-ProfitMarginDatabaseConnectorPipe" }, { "// ": "------------- End of dataflow -------------", "// The connector dataflow terminates here by generating a log summary through the Logging Service": "", "name": "Terminal", "label": "Terminal" } ] }

Pulse Ejecutar.

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 Docker.
  1. docker exec -it commerce_registry_1 bash
  2. /opt/nifi-registry/scripts/export_flow.sh ProfitMarginSchemaUpdateConnector>/opt/nifi-registry/flows/ProfitMarginSchemaUpdateConnector.json
  3. /opt/nifi-registry/scripts/export_flow.sh ProfitMarginDatabaseConnectorPipe>/opt/nifi-registry/flows/ProfitMarginDatabaseConnectorPipe.json
  4. exit
  5. docker cp commerce_registry_1:/opt/nifi-registry/flows/ProfitMarginDatabaseConnectorPipe.json ./ProfitMarginDatabaseConnectorPipe.json
  6. 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.