Configurer le connecteur dans NiFi

Dans cette leçon, vous configurez le connecteur que vous avez créé à l'étape précédente à l'aide de NiFi.

Before you begin

Téléchargez et extrayez le fichier SampleVersionedConnectors9112.zip dans un répertoire temporaire dans votre environnement de développement. Ce fichier contient un exemple de fichier JSON pour les connecteurs par version.

About this task

Utilisez le lien suivant pour NiFi : http://hostname/IP:30600/nifi/

Note:

Cette leçon utilise le nom du connecteur custom.cas. Veillez à remplacer custom.cas par le nom du connecteur que vous avez créé à l'étape précédente.
Si vous êtes sur la page d'accueil de NiFi où vous pouvez afficher tous les processeurs, vous devrez peut-être faire défiler l'écran vers la droite du pipeline et effectuer un zoom avant pour trouver les processeurs requis.

Procedure

Pour configurer le connecteur dans NiFi, procédez comme suit.

Faites défiler l'écran jusqu'au côté droit de la zone de conception NiFi. Le connecteur custom.cas que vous avez créé à l'étape précédente se trouve ici.
Mettez à jour les propriétés du processeur custom-cas_Template-Schema-CAS.
1. Allez au nouveau pipeline custom.cas et cliquez deux fois sur custom-_Template-Schema-CAS.
2. Cliquez deux fois sur le groupe de processus Mise à jour du schéma.
3. Recherchez le groupe de processus Configurer le schéma d'index Elasticsearch et double-cliquez dessus.
4. Recherchez et cliquez avec le bouton droit de la souris sur Spécifier le schéma à mettre à jour et arrêtez le processeur.
5. Cliquez deux fois sur le processeur pour modifier les paramètres.
6. Allez dans l'onglet Propriétés et mettez à jour la propriété param.schema avec la valeur product.
7. Cliquez sur Appliquer.
8. Démarrez le processeur en cliquant avec le bouton droit de la souris et en sélectionnant Démarrer.

Mettez à jour le processeur Renseigner le schéma d'index.

Arrêtez le processeur.
Cliquez deux fois sur le processeur pour modifier les paramètres.

Mettez à jour la propriété Valeur de remplacement avec la valeur suivante :


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

Démarrez le processeur.

Ouvrez Flux NiFi > custom.cas - _Template-DatabaseETL-CAS > Canal du connecteur personnalisé > Processeur du canal du connecteur personnalisé
Faites glisser un bouton Processeur vers un espace vide dans la zone de conception, puis entrez Personnalisé. Sélectionnez CustomFieldDocumentProcessor et cliquez sur le bouton AJOUTER.
Ajoutez customFieldDocumentProcessor, comme dans l'exemple de capture d'écran suivant.
Note: Vous devez arrêter les processeurs correspondants pour modifier les relations de flux.
Sélectionnez l'onglet de propriétés CustomFieldDocumentProcessor et renseignez les valeurs des propriétés suivantes :
- Database Connection Pooling Service: Database Connection Pool
- SQL Pre-Query: ${ingest.database.schema.sql}
- SQL select query: ${ingest.database.sql}

Ouvrez Définir le processeur de SQL personnalisé. Mettez à jour le processeur Définir le SQL personnalisé et le nom d'index. Mettez à jour ingest.database.sql avec la SQL ci-dessous :

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})

Notez la clause OP.CATENTRY_ID IN (${data.catentryId}) dans la SQL. Chaque flux traitera un lot d'entrées de catalogue.

Le résultat de cette SQL contient deux colonnes, CATENTRY_ID et PROFIT_MARGIN.

Note: Assurez-vous que les données de prix de revient sont disponibles dans votre base de données. Ce connecteur ne vérifie pas si le SQL renvoie des résultats. Si les données de prix de revient ne sont pas disponibles dans la base de données, les données renvoyées pour le SQL sont vides et entraînent des erreurs.

Créez une version pour custom-cas_Template-Schema-CAS.
Cliquez avec le bouton droit de la souris sur le groupe de processus custom-cas_Template-Schema-CAS.
Cliquez sur Version > Arrêter le contrôle des versions > Déconnecter.
Cliquez avec le bouton droit de la souris sur le groupe de processus custom-cas_Template-Schema-CAS.
Cliquez sur Version > > Démarrer le contrôle des versions.
Ajoutez le Flow Name requis et enregistrez.
Dans ce tutoriel, le Flow Name = ProfitMarginSchemaUpdateConnector.
Créez une version pour custom.cas - _Template-DatabaseETL-CAS.

Cliquez avec le bouton droit de la souris sur le groupe de processus, custom.cas - _Template-DatabaseETL-CAS.

Cliquez sur Version > Arrêter le contrôle des versions > Déconnecter.

Cliquez avec le bouton droit de la souris sur le groupe de processus, custom.cas - _Template-DatabaseETL-CAS.

Cliquez sur Version > > Arrêter le contrôle des versions.

Ajoutez le nom de flux requis et enregistrez.

Dans ce tutoriel, le Flow Name = ProfitMarginDatabaseConnectorPipe.
Copiez le groupe de processus custom.cas - _Template-Schema-CAS et collez-le dans un espace vide en regard de auth.reindex.cas - Schéma d'espace de travail.

Insérez ce nouveau groupe de processus personnalisé entre le schéma d'espace de travail et le groupe de processus suivant comme suit :

La séquence du groupe de processus peut changer. Le groupe de processus personnalisé doit être inséré après le schéma d'espace de travail.

Cliquez avec le bouton droit de la souris sur le groupe de processus custom.cas - _Template-Schema-CAS et sélectionnez Démarrer pour démarrer le groupe de processus.
Procédez comme suit pour ajouter le groupe de processus _Template-DatabaseETL-CAS avant le groupe de processus Envoyer le produit à Elasticsearch.
Cliquez avec le bouton droit de la souris sur le groupe de processus _Template-DatabaseETL-CAS et sélectionnez Démarrer pour démarrer le groupe de processus.
Note: Si vous voyez un composant non valide dans le groupe _custom.cas - _Template-DatabaseETL-CAS _process, procédez comme suit :
Cliquez avec le bouton droit de la souris sur custom.cas - _Template-DatabaseETL-CAS, puis sélectionnez Configurer.

Accédez à Services du contrôleur.

Cliquez sur Activer pour les services Pool de connexions à la base de données.
- Cliquez sur le bouton Activer et, pour quitter la fenêtre en incrustation Activer les services du contrôleur, cliquez sur Fermer.
- Cliquez avec le bouton droit de la souris sur le groupe de processus custom.cas - _Template- DatabaseETL-CAS et sélectionnez Démarrer.
Extraye le connecteur par version.
1. Exécutez les commandes suivantes :
  Note: Le nom du conteneur Docker utilisé dans cet exemple est commerce_registry_1. Remplacez-le par le nom réel de votre conteneur 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
```
  Vous pouvez également extraire le canal du connecteur via l'interface NiFi Registry.
  1. Lancez l'interface NiFi Registry et localisez le nom de flux que vous souhaitez exporter, par exemple ProfitMarginSchemaUpdateConnector ou ProfitMarginDatabaseConnectorPipe.
  2. Cliquez sur le bouton Action et sélectionnez Exporter la version.
  3. Sélectionnez la version la plus récente et cliquez sur Exporter.
2. Copiez ProfitMarginSchemaUpdateConnector.json et ProfitMarginDatabaseConnectorPipe.json dans votre environnement d'exécution.