Création et configuration de pipelines

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

Les options permettant de créer et de configurer un pipeline sont les suivantes :

Charger le pipeline :
- Reportez-vous à Création et configuration d'un pipeline pour le chargement de données.
Pipeline d'export :
- Exportez les résultats incrémentiels d'une requête vers la banque d'objets en utilisant une colonne de date ou d'horodatage comme clé pour le suivi des données plus récentes. Reportez-vous à Création et configuration d'un pipeline pour l'export avec colonne d'horodatage.
- Exportez les données incrémentielles d'une table vers la banque d'objets en utilisant une colonne de date ou d'horodatage comme clé pour le suivi des données plus récentes. Reportez-vous à Création et configuration d'un pipeline pour l'export avec colonne d'horodatage.
- Exportez les données d'une table vers une banque d'objets à l'aide d'une requête pour sélectionner des données sans référence à une colonne de date ou d'horodatage (de sorte que le pipeline exporte toutes les données que la requête sélectionne pour chaque exécution du planificateur). Reportez-vous à Création et configuration d'un pipeline pour exporter des résultats de requête (sans horodatage).

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 une banque d'objets ou des répertoires vers des tables dans la base de données Autonomous AI.

Un pipeline de chargement consomme des données placées sur une banque d'objets ou dans des répertoires et les charge dans une table de la base de données Autonomous AI. Lorsque vous créez un pipeline de chargement, ce dernier est exécuté à intervalles réguliers pour utiliser les données placées sur l'emplacement source, 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 l'emplacement de la source 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 base de données Autonomous AI, utilisez une table existante ou créez la table de base de données dans laquelle vous chargez des données. Par exemple :

CREATE TABLE EMPLOYEE
            (name     VARCHAR2(128),
             age      NUMBER,
             salary   NUMBER);

Créez un pipeline pour charger des données à partir d'objets de banque d'objets ou d'objets de répertoire.

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.

Créez un objet d'informations d'identification pour accéder à la banque d'objets contenant 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 ne fournissez pas de valeur 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.

Définissez les attributs de pipeline, y compris les attributs requis : location, table_name et format.

Cas n°1 : créez un pipeline pour charger des données à partir de la banque d'objets.

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;
/

Cas 2 : créez un pipeline pour charger des données à partir d'objets de répertoire.

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       pipeline_type => 'LOAD',
       attributes    => JSON_OBJECT(
            'location' VALUE 'MY_DIR:*.csv',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"csv"}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

Cas 3 : Créez un pipeline pour charger des fichiers CSV à partir d'Amazon S3 à l'aide de caractères génériques.

Utilisez une URL Amazon S3 avec des informations d'identification AWS lorsque les fichiers source sont stockés dans Amazon S3. Cet exemple utilise un caractère générique dans l'emplacement de la banque d'objets pour charger uniquement les fichiers qui correspondent au modèle.

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       attributes    => JSON_OBJECT(
            'credential_name' VALUE 'AWS_CRED_NAME',
            'location' VALUE 'https://aws-bucket-01.s3.amazonaws.com/incoming/employee_*.csv',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"csv", "delimiter":",", "header":true}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

Cas 4 : créez un pipeline pour charger les fichiers Parquet à partir d'OCI Object Storage.

Utilisez une URL OCI Object Storage avec des informations d'identification OCI lorsque les fichiers source sont stockés dans un bucket OCI. Cet exemple montre comment charger des fichiers Parquet à partir d'un préfixe de bucket.

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       attributes    => JSON_OBJECT(
            'credential_name' VALUE 'OCI_CRED_NAME',
            'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/employee/',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"parquet", "schema":"first"}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

Cas 5 : créez un pipeline pour charger les fichiers Oracle Data Pump à partir de la banque d'objets.

Utilisez le format datapump lorsque les fichiers source sont des fichiers dump Oracle Data Pump. Cet exemple montre comment charger des fichiers dump à partir d'OCI Object Storage.

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       attributes    => JSON_OBJECT(
            'credential_name' VALUE 'OCI_CRED_NAME',
            'location' VALUE 'https://objectstorage.us-phoenix-1.oraclecloud.com/n/namespace-string/b/bucketname/o/datapump/employee_%U.dmp',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"datapump"}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

Cas 6 : créez un pipeline pour charger des fichiers CSV à partir d'Azure Blob Storage.

Utilisez une URL de stockage BLOB Azure avec des informations d'identification de principal de service Azure lorsque les fichiers source sont stockés dans le stockage BLOB Azure.

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'MY_PIPE1',
       attributes    => JSON_OBJECT(
            'credential_name' VALUE 'AZURE$PA',
            'location' VALUE 'https://azurestorageaccount.blob.core.windows.net/container/employee_*.csv',
            'table_name' VALUE 'employee',
            'format' VALUE '{"type":"csv", "delimiter":",", "header":true}',
            'priority' VALUE 'HIGH',
            'interval' VALUE '20')
  );
END;
/

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

location : indique l'emplacement du fichier source sur l'objet de banque d'objets ou de répertoire.
table_name : indique la table de la base de données dans laquelle vous chargez les données. La valeur location que vous indiquez concerne une valeur table_name par pipeline.
format : décrit le format des données que vous chargez.

Pour plus d'informations, reportez-vous à Options de format du package DBMS_CLOUD et Options de format du package DBMS_CLOUD pour Avro, ORC ou Parquet

credential_name correspond aux 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 exécute chaque exécution plus rapidement, par rapport à une exécution avec une priorité plus faible.

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

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

Après avoir créé un pipeline, vous pouvez le tester ou le démarrer :
- Tester les pipelines
- Démarrer un pipeline

Vous pouvez également utiliser le format suivant pour définir le format JSON :

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 temporelle de votre base de données Autonomous AI vers la banque d'objets.

A l'aide de cette option de pipeline d'export, vous indiquez une table ou une requête SQL et une colonne avec un horodatage que le pipeline utilise pour suivre l'heure du dernier téléchargement. Vous pouvez utiliser un pipeline d'export pour partager des données à des fins de consommation 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 des données de votre base de données Autonomous AI vers la banque d'objets. Lorsque vous créez un pipeline d'export, ce dernier est exécuté à intervalles réguliers et place les données dans la banque d'objets.

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.

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

Vous indiquez les informations d'identification de l'emplacement de destination de pipeline avec l'attribut credential_name. Si vous ne fournissez pas de valeur 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.

Définissez les attributs de 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 de sorte que seules les données requises soient exportées vers la banque d'objets.

Utilisation d'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;
/

Utilisation 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;
/

Exportez les résultats de requête sous forme de fichiers CSV vers OCI Object Storage avec des noms de colonne dans la ligne d'en-tête :

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

Exporter les données de table en tant que fichiers Parquet vers Amazon S3 :

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'EXP_PIPE1',
       attributes    => JSON_OBJECT(
          'credential_name' VALUE 'AWS_CRED_NAME',
          'location' VALUE 'https://aws-bucket-01.s3.amazonaws.com/exports/',
          'table_name' VALUE 'metrics_table',
          'key_column' VALUE 'metric_time',
          'format' VALUE '{"type":"parquet"}',
          'priority' VALUE 'MEDIUM',
          'interval' VALUE '20')
    );
END;
/

Exportez les résultats des requêtes en tant que fichiers Oracle Data Pump vers la banque d'objets :

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

Exporter les résultats de requête sous forme de fichiers JSON vers le stockage BLOB Azure :

BEGIN
     DBMS_CLOUD_PIPELINE.SET_ATTRIBUTE(
       pipeline_name => 'EXP_PIPE1',
       attributes    => JSON_OBJECT(
          'credential_name' VALUE 'AZURE$PA',
          'location' VALUE 'https://azurestorageaccount.blob.core.windows.net/container/exports/metrics/',
          'query' VALUE 'SELECT * FROM metrics_table',
          '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 concerne une valeur table_name par pipeline.
table_name : indique la table de la base de données qui contient les données que vous exportez. Indiquez le paramètre table_name ou le paramètre query.
query : indique la requête à exécuter dans la base de données qui fournit les données que vous exportez. Indiquez le paramètre table_name ou le paramètre query.
format : décrit le format des données que vous exportez.

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

Pour les pipelines d'export, le type de format suit les types de fichier de sortie pris en charge par DBMS_CLOUD.EXPORT_DATA : CSV, JSON, Parquet, XML et Oracle Data Pump.

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 de temps en minutes entre les exécutions consécutives d'un travail de pipeline. L'intervalle interval par défaut est de 15 minutes.

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

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 base de données Autonomous AI 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 dernières données de votre base de données Autonomous AI vers la banque d'objets afin que d'autres applications puissent les utiliser.

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.

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

Vous indiquez les informations d'identification de l'emplacement de destination de pipeline avec l'attribut credential_name. Si vous ne fournissez pas de valeur 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.
Définissez les attributs de 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 à Options de format de package DBMS_CLOUD 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 de temps en minutes entre les exécutions consécutives d'un travail de pipeline. L'intervalle interval par défaut est de 15 minutes.

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

Après avoir créé un pipeline, vous pouvez le tester ou le démarrer :
- Tester les pipelines
- Démarrer un pipeline

Documentation Oracle Cloud Infrastructure

Création et configuration de pipelines

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

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

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