Création et configuration de pipelines

Vous pouvez créer des pipelines de chargement ou d'export. Lorsque vous créez un pipeline, vous utilisez des paramètres et définissez des attributs de pipeline pour le configurer.

Les options de création et de configuration d'un pipeline sont les suivantes :

Création et configuration d'un pipeline pour le chargement de données

Vous pouvez créer un pipeline pour charger des données à partir de fichiers externes dans la banque d'objets vers des tables dans Autonomous Database.

Un pipeline de chargement consomme les données placées sur la banque d'objets et les charge vers une table dans Autonomous Database. Lorsque vous créez un pipeline de chargement, il est exécuté à intervalles réguliers pour consommer les données placées dans la banque d'objets. Lorsque de nouveaux fichiers de données arrivent, le pipeline charge les nouvelles données. Vous pouvez également utiliser un pipeline pour copier de manière fiable des fichiers, avec des fonctionnalités de reprise et de nouvelle tentative, de la banque d'objets vers une table de la base de données.

Avec un pipeline de chargement, le package de pipeline utilise DBMS_CLOUD.COPY_DATA pour charger les données.

Sur votre instance Autonomous Database, utilisez une table existante ou créez la table de base de données dans laquelle vous chargez les données. Exemple : 

CREATE TABLE EMPLOYEE
            (name     VARCHAR2(128),
             age      NUMBER,
             salary   NUMBER);
  1. Créez un pipeline pour charger des données à partir de la banque d'objets.
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name => 'MY_PIPE1',
        pipeline_type => 'LOAD',
        description   => 'Load metrics from object store into a table'
  );
END;
/

    Pour plus d'informations, reportez-vous à Procédure CREATE_PIPELINE.

  2. Créez un objet d'informations d'identification pour accéder à la banque d'objets qui contient les fichiers que vous chargez.

    Vous indiquez les informations d'identification de l'emplacement de source de pipeline avec l'attribut credential_name. Si vous n'indiquez pas de credential_name à l'étape suivante, la valeur credential_name est définie sur NULL. Vous pouvez utiliser la valeur NULL par défaut lorsque l'attribut location est une URL publique ou pré-authentifiée.

    Pour plus d'informations, reportez-vous à Procédure CREATE_CREDENTIAL.

  3. Définissez les attributs de pipeline, y compris les attributs requis : location, table_name et format.
    BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       attributes    => JSON_OBJECT(
            'credential_name' VALUE 'OBJECT_STORE_CRED',
            'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"json", "columnpath":["$.NAME", "$.AGE", "$.SALARY"]}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

    Les attributs suivants doivent être définis pour exécuter un pipeline de chargement :

    • location : indique l'emplacement du fichier source dans la banque d'objets.

    • table_name : indique la table de la base de données dans laquelle vous chargez les données. La valeur location que vous indiquez correspond à une valeur table_name par pipeline.

    • format : décrit le format des données en cours de chargement.

      Pour plus d'informations, reportez-vous à Options de format de package DBMS_CLOUD.

    credential_name est les informations d'identification que vous avez créées à l'étape précédente.

    La valeur priority détermine le nombre de fichiers chargés en parallèle. Un pipeline avec une priorité plus élevée consomme plus de ressources de base de données et termine chaque exécution plus rapidement, par rapport à une exécution avec une priorité moins élevée.

    La valeur interval indique l'intervalle en minutes entre les exécutions consécutives d'un travail de pipeline. La valeur par défaut de interval est de 15 minutes.

    Pour plus d'informations sur les attributs de pipeline, reportez-vous à DBMS_CLOUD_PIPELINE Attributes.

    Après avoir créé un pipeline, vous pouvez le tester ou le démarrer :

Sinon, pour définir le format pour JSON, vous pouvez utiliser le format suivant :

BEGIN
    DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
        pipeline_name   => 'MY_PIPE1',
        attribute_name  => 'format',
        attribute_value => JSON_OBJECT('type' value 'json', 'columnpath' value '["$.NAME", "$.AGE", "$.SALARY"]')
    );
END;
/

Création et configuration d'un pipeline pour l'export avec une colonne d'horodatage

Vous pouvez créer un pipeline d'export pour exporter automatiquement les données de série chronologique de votre instance Autonomous Database vers la banque d'objets.

A l'aide de cette option de pipeline d'export, vous indiquez une table ou une requête SQL, ainsi qu'une colonne avec un horodatage que le pipeline utilise pour effectuer le suivi de l'heure du dernier téléchargement. Vous pouvez utiliser un pipeline d'export pour partager des données destinées à être utilisées par d'autres applications ou pour enregistrer des données dans la banque d'objets.

Avec un pipeline d'export, le package de pipeline utilise DBMS_CLOUD.EXPORT_DATA pour exporter des données.

Un pipeline d'export exporte les données de votre instance Autonomous Database vers la banque d'objets. Lorsque vous créez un pipeline d'export, il s'exécute à intervalles réguliers et place les données dans la banque d'objets.

  1. Créez un pipeline pour exporter des données vers la banque d'objets.
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name=>'EXP_PIPE1',
        pipeline_type=>'EXPORT',
        description=>'Export time series metrics to object store');
END;
/

    Pour plus d'informations, reportez-vous à Procédure CREATE_PIPELINE.

  2. Créez un objet d'informations d'identification pour accéder à l'emplacement de banque d'objets de destination où vous exportez les fichiers de données.

    Vous indiquez les informations d'identification pour l'emplacement de destination de pipeline avec l'attribut credential_name. Si vous n'indiquez pas de credential_name à l'étape suivante, la valeur credential_name est définie sur NULL. Vous pouvez utiliser la valeur NULL par défaut lorsque l'attribut location est une URL publique ou pré-authentifiée.

    Pour plus d'informations, reportez-vous à Procédure CREATE_CREDENTIAL.

  3. Définissez les attributs du pipeline d'export.

    Lorsque vous indiquez un paramètre table_name, les lignes de table sont exportées vers la banque d'objets. Lorsque vous indiquez un paramètre query, la requête indique une instruction SELECT afin que seules les données requises soient exportées vers la banque d'objets.

    • En utilisant un paramètre table_name : 

      BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'EXP_PIPE1',
       attributes    => JSON_OBJECT('credential_name' VALUE 'OBJECT_STORE_CRED',
          'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/',
          'table_name' VALUE 'metric_table',
          'key_column' VALUE 'metric_time',
          'format' VALUE '{"type": "json"}',
          'priority' VALUE 'MEDIUM',
          'interval' VALUE '20')
  );
END;
/

    • A l'aide d'un paramètre query : 

      BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'EXP_PIPE1',
       attributes    => JSON_OBJECT('credential_name' VALUE 'OBJECT_STORE_CRED',
           'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/',
           'query' VALUE 'SELECT * from metrics_table',
           'key_column' VALUE 'metric_time',
           'format' VALUE '{"type": "json"}',
           'priority' VALUE 'MEDIUM',
           'interval' VALUE '20')
  );
END;
/

    Où credential_name correspond aux informations d'identification que vous avez créées à l'étape précédente.

    Les attributs suivants doivent être définis pour exécuter un pipeline d'export :

    • location : indique l'emplacement de la banque d'objets de destination. La valeur location que vous indiquez correspond à une valeur table_name par pipeline.

    • table_name : indique la table de la base de données contenant les données que vous exportez (le paramètre table_name ou le paramètre query est requis).

    • query : indique la requête à exécuter dans la base de données qui fournit les données que vous exportez (le paramètre table_name ou le paramètre query est requis).

    • format : décrit le format des données que vous exportez.

      Pour plus d'informations, reportez-vous à DBMS_CLOUD Options de format de package pour EXPORT_DATA.

    La valeur priority détermine le degré de parallélisme pour l'extraction des données de la base de données.

    La valeur interval indique l'intervalle en minutes entre les exécutions consécutives d'un travail de pipeline. La valeur par défaut de interval est de 15 minutes.

    Pour plus d'informations sur les attributs de pipeline, reportez-vous à DBMS_CLOUD_PIPELINE Attributes.

    Après avoir créé un pipeline, vous pouvez le tester ou le démarrer :

Création et configuration d'un pipeline pour exporter des résultats de requête (sans horodatage)

Vous pouvez créer un pipeline d'export pour exporter automatiquement les données de votre instance Autonomous Database vers la banque d'objets. A l'aide de cette option de pipeline d'export, vous indiquez une requête SQL que le pipeline exécute régulièrement pour exporter des données vers la banque d'objets. Vous pouvez utiliser cette option d'export pour partager les données les plus récentes de votre instance Autonomous Database vers la banque d'objets pour que d'autres applications utilisent les données.

Un pipeline d'export exporte les données de votre instance Autonomous Database vers la banque d'objets. Lorsque vous créez un pipeline d'export, il s'exécute à intervalles réguliers et place les données dans la banque d'objets.

  1. Créez un pipeline pour exporter des données vers la banque d'objets.
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name=>'EXP_PIPE2',
        pipeline_type=>'EXPORT',
        description=>'Export query results to object store.');
END;
/

    Pour plus d'informations, reportez-vous à Procédure CREATE_PIPELINE.

  2. Créez un objet d'informations d'identification pour accéder à l'emplacement de banque d'objets de destination où vous exportez les fichiers de données.

    Vous indiquez les informations d'identification pour l'emplacement de destination de pipeline avec l'attribut credential_name. Si vous n'indiquez pas de credential_name à l'étape suivante, la valeur credential_name est définie sur NULL. Vous pouvez utiliser la valeur NULL par défaut lorsque l'attribut location est une URL publique ou pré-authentifiée.

    Pour plus d'informations, reportez-vous à Procédure CREATE_CREDENTIAL.

  3. Définissez les attributs du pipeline d'export.
    BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'EXP_PIPE2',
       attributes    => JSON_OBJECT(
          'credential_name' VALUE 'OBJECT_STORE_CRED',
          'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/',
          'query' VALUE 'SELECT * FROM table_name',
          'format' VALUE '{"type": "json"}',
          'priority' VALUE 'MEDIUM',
          'interval' VALUE '20')
  );
END;
/

    Où credential_name correspond aux informations d'identification que vous avez créées à l'étape précédente.

    Les attributs suivants doivent être définis pour exécuter un pipeline d'export :

    • location : indique l'emplacement de la banque d'objets de destination.

    • query : indique la requête à exécuter dans la base de données qui fournit les données que vous exportez.

    • format : décrit le format des données que vous exportez.

      Pour plus d'informations, reportez-vous à DBMS_CLOUD Options de format de package pour EXPORT_DATA.

    La valeur priority détermine le degré de parallélisme pour l'extraction des données de la base de données.

    La valeur interval indique l'intervalle en minutes entre les exécutions consécutives d'un travail de pipeline. La valeur par défaut de interval est de 15 minutes.

    Pour plus d'informations sur les attributs de pipeline, reportez-vous à DBMS_CLOUD_PIPELINE Attributes.

    Après avoir créé un pipeline, vous pouvez le tester ou le démarrer :

Tester les pipelines

Utilisez RUN_PIPELINE_ONCE pour exécuter un pipeline une fois à la demande sans créer de travail programmé.

RUN_PIPELINE_ONCE est utile pour tester un pipeline avant de le démarrer. Une fois que vous avez exécuté un pipeline pour tester le pipeline et vérifier qu'il fonctionne comme prévu, utilisez RESET_PIPELINE pour réinitialiser l'état du pipeline (à l'état précédant l'exécution de RUN_PIPELINE_ONCE).

  1. Création d'un pipeline.
  2. Exécutez un pipeline une fois pour le tester.
    BEGIN
    DBMS_CLOUD_PIPELINE.RUN_PIPELINE_ONCE(
        pipeline_name => 'MY_PIPE1'
);
END;
/

    Pour plus d'informations, reportez-vous à Procédure RUN_PIPELINE_ONCE.

  3. Effectuez toutes les vérifications requises pour vérifier que le pipeline fonctionne comme prévu.

    Pour plus d'informations, reportez-vous à Surveillance et dépannage des pipelines.

  4. Réinitialisez le pipeline.
    BEGIN  
   DBMS_CLOUD_PIPELINE.RESET_PIPELINE(
     pipeline_name => 'MY_PIPE1',
     purge_data => TRUE
);
END;
/

    Pour plus d'informations, reportez-vous à Procédure RESET_PIPELINE.