Crea e configura pipeline

È possibile creare una o più pipeline di caricamento o esportazione. Quando si crea una pipeline, è possibile utilizzare i parametri e impostare gli attributi della pipeline per configurare la pipeline.

Di seguito sono riportate le opzioni per creare e configurare una pipeline.

• Pipeline di caricamento:

  • Vedere Creazione e configurazione di una pipeline per il caricamento dei dati.

• Esporta pipeline:

  • Esporta i risultati incrementali di una query nell'area di memorizzazione degli oggetti utilizzando una colonna data o indicatore orario come chiave per tenere traccia dei dati più recenti. Vedere Creazione e configurazione di una pipeline per l'esportazione con colonna indicatore orario.

  • Esportare i dati incrementali di una tabella nell'area di memorizzazione degli oggetti utilizzando una colonna data o indicatore orario come chiave per la registrazione dei dati più recenti. Vedere Creazione e configurazione di una pipeline per l'esportazione con colonna indicatore orario.

  • Esportare i dati di una tabella nell'area di memorizzazione degli oggetti utilizzando una query per selezionare i dati senza un riferimento a una colonna data o indicatore orario (in modo che la pipeline esporta tutti i dati selezionati dalla query per ogni esecuzione dello scheduler). Vedere Creazione e configurazione di una pipeline per l'esportazione dei risultati delle query (senza indicatore orario).

Creare e configurare una pipeline per il caricamento dei dati

È possibile creare una pipeline per caricare i dati da file esterni nell'area di memorizzazione degli oggetti o nelle directory nelle tabelle in Autonomous AI Database.

Una pipeline di caricamento utilizza i dati inseriti nell'area di memorizzazione degli oggetti o nelle directory e li carica in una tabella in Autonomous AI Database. Quando si crea una pipeline di caricamento, la pipeline viene eseguita a intervalli regolari per utilizzare i dati inseriti nella posizione di origine, quando arrivano nuovi file di dati la pipeline carica i nuovi dati. È inoltre possibile utilizzare una pipeline per copiare in modo affidabile i file, con funzionalità di ripresa e nuovo tentativo, dalla posizione di origine a una tabella del database.

Con una pipeline di caricamento, il package della pipeline utilizza DBMS_CLOUD.COPY_DATA per caricare i dati.

In Autonomous AI Database, utilizzare una tabella esistente o creare la tabella di database in cui si stanno caricando i dati. Ad esempio:

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

1. Creare una pipeline per caricare i dati dall'area di memorizzazione degli oggetti o dagli oggetti directory.

     BEGIN
      DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
         pipeline_name => 'MY_PIPE1',
         pipeline_type => 'LOAD',
         description   => 'Load metrics from object store into a table'
      );
     END;
    /
   

   Per ulteriori informazioni, vedere Procedura CREATE_PIPELINE.

2. Creare un oggetto credenziale per accedere all'area di memorizzazione degli oggetti che contiene i file che si stanno caricando.

   Specificare la credenziale per la posizione di origine della pipeline con l'attributo credential_name. Se non si specifica un valore credential_name nel passo successivo, il valore credential_name viene impostato su NULL. È possibile utilizzare il valore NULL predefinito quando l'attributo location è un URL pubblico o preautenticato.

   Per ulteriori informazioni, vedere Procedura CREDENTIAL.

3. Impostare gli attributi della pipeline, inclusi gli attributi obbligatori: location, table_name e format.

   Caso 1: creare una pipeline per il caricamento dei dati dall'area di memorizzazione degli oggetti.

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

   Caso 2: creare una pipeline per il caricamento dei dati dagli oggetti directory.

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

   Per eseguire una pipeline di caricamento in entrambi i casi, è necessario impostare i seguenti attributi:

   • location: specifica la posizione del file di origine nell'area di memorizzazione degli oggetti o nell'oggetto directory.

   • table_name: specifica la tabella nel database in cui si stanno caricando i dati. Il valore location specificato si riferisce a un valore table_name per pipeline.

   • format: descrive il formato dei dati da caricare.

     Per ulteriori informazioni, vedere DBMS_CLOUD Package Format Options.

   credential_name è la credenziale creata nel passo precedente.

   Il valore priority determina il numero di file caricati in parallelo. Una pipeline con priorità più elevata utilizza un numero maggiore di risorse del database e completa ciascuna esecuzione più velocemente rispetto all'esecuzione con priorità inferiore.

   Il valore interval specifica l'intervallo di tempo in minuti tra le esecuzioni consecutive di un job pipeline. L'intervallo predefinito è interval di 15 minuti.

   Per informazioni dettagliate sugli attributi della pipeline, vedere Attributi DBMS_CLOUD_PIPELINE.

4. Dopo aver creato una pipeline, è possibile eseguire il test della pipeline o avviarla:

   • Esegui test pipeline

   • Avvia una pipeline

In alternativa, per impostare il formato per JSON, è possibile utilizzare il seguente formato:

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

Crea e configura una pipeline per l'esportazione con colonna indicatore orario

È possibile creare una pipeline di esportazione per esportare automaticamente i dati delle serie temporali da Autonomous AI Database all'area di memorizzazione degli oggetti.

Utilizzando questa opzione di pipeline di esportazione è possibile specificare una tabella o una query SQL e una colonna con un indicatore orario utilizzato dalla pipeline per tenere traccia dell'ora dell'ultimo caricamento. È possibile utilizzare una pipeline di esportazione per condividere i dati per il consumo da parte di altre applicazioni o per salvare i dati nell'area di memorizzazione degli oggetti.

Con una pipeline di esportazione, il package pipeline utilizza DBMS_CLOUD.EXPORT_DATA per esportare i dati.

Una pipeline di esportazione esporta i dati dall'Autonomous AI Database nell'area di memorizzazione degli oggetti. Quando si crea una pipeline di esportazione, la pipeline viene eseguita a intervalli regolari e inserisce i dati nell'area di memorizzazione degli oggetti.

1. Creare una pipeline per esportare i dati nell'area di memorizzazione degli oggetti.

    BEGIN
         DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
            pipeline_name=>'EXP_PIPE1',
            pipeline_type=>'EXPORT',
            description=>'Export time series metrics to object store');
    END;
    /
   

   Per ulteriori informazioni, vedere Procedura CREATE_PIPELINE.

2. Creare un oggetto credenziale per accedere alla posizione dell'area di memorizzazione degli oggetti di destinazione in cui si esportano i file di dati.

   Specificare la credenziale per la posizione di destinazione della pipeline con l'attributo credential_name. Se non si specifica un valore credential_name nel passo successivo, il valore credential_name viene impostato su NULL. È possibile utilizzare il valore NULL predefinito quando l'attributo location è un URL pubblico o preautenticato.

   Per ulteriori informazioni, vedere Procedura CREDENTIAL.

3. Impostare gli attributi della pipeline di esportazione.

   Quando si specifica un parametro table_name, le righe di tabella vengono esportate nell'area di memorizzazione degli oggetti. Quando si specifica un parametro query, la query specifica un'istruzione SELECT in modo che solo i dati richiesti vengano esportati nell'area di memorizzazione degli oggetti.

   • Utilizzando un parametro 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;
     /
     

   • Utilizzando un parametro 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;
     /
     

     Dove credential_name è la credenziale creata nel passo precedente.

   Per eseguire una pipeline di esportazione è necessario impostare i seguenti attributi:

   • location: specifica la posizione dell'area di memorizzazione degli oggetti di destinazione. Il valore location specificato si riferisce a un valore table_name per pipeline.

   • table_name: specifica la tabella del database contenente i dati che si stanno esportando (è necessario il parametro table_name o il parametro query).

   • query: specifica la query da eseguire nel database che fornisce i dati che si stanno esportando (è necessario il parametro table_name o il parametro query).

   • format: descrive il formato dei dati che si stanno esportando.

     Per ulteriori informazioni, vedere DBMS_CLOUD Package Format Options for EXPORT_DATA.

   Il valore priority determina il grado di parallelismo per il recupero dei dati dal database.

   Il valore interval specifica l'intervallo di tempo in minuti tra le esecuzioni consecutive di un job pipeline. L'intervallo predefinito è interval di 15 minuti.

   Per informazioni dettagliate sugli attributi della pipeline, vedere Attributi DBMS_CLOUD_PIPELINE.

   Dopo aver creato una pipeline, è possibile eseguire il test della pipeline o avviarla:

   • Esegui test pipeline

   • Avvia una pipeline

Creazione e configurazione di una pipeline per l'esportazione dei risultati della query (senza indicatore orario)

È possibile creare una pipeline di esportazione per esportare automaticamente i dati da Autonomous AI Database nell'area di memorizzazione degli oggetti. Questa opzione di pipeline di esportazione consente di specificare una query SQL che la pipeline esegue periodicamente per esportare i dati nell'area di memorizzazione degli oggetti. È possibile utilizzare questa opzione di esportazione per condividere i dati più recenti da Autonomous AI Database all'area di memorizzazione degli oggetti in modo che altre applicazioni utilizzino i dati.

Una pipeline di esportazione esporta i dati dall'Autonomous AI Database nell'area di memorizzazione degli oggetti. Quando si crea una pipeline di esportazione, la pipeline viene eseguita a intervalli regolari e inserisce i dati nell'area di memorizzazione degli oggetti.

1. Creare una pipeline per esportare i dati nell'area di memorizzazione degli oggetti.

    BEGIN
         DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
            pipeline_name=>'EXP_PIPE2',
            pipeline_type=>'EXPORT',
            description=>'Export query results to object store.');
    END;
    /
   

   Per ulteriori informazioni, vedere Procedura CREATE_PIPELINE.

2. Creare un oggetto credenziale per accedere alla posizione dell'area di memorizzazione degli oggetti di destinazione in cui si esportano i file di dati.

   Specificare la credenziale per la posizione di destinazione della pipeline con l'attributo credential_name. Se non si specifica un valore credential_name nel passo successivo, il valore credential_name viene impostato su NULL. È possibile utilizzare il valore NULL predefinito quando l'attributo location è un URL pubblico o preautenticato.

   Per ulteriori informazioni, vedere Procedura CREDENTIAL.

3. Impostare gli attributi della pipeline di esportazione.

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

   Dove credential_name è la credenziale creata nel passo precedente.

   Per eseguire una pipeline di esportazione è necessario impostare i seguenti attributi:

   • location: specifica la posizione dell'area di memorizzazione degli oggetti di destinazione.

   • query: specifica la query da eseguire nel database che fornisce i dati da esportare.

   • format: descrive il formato dei dati che si stanno esportando.

     Per ulteriori informazioni, vedere DBMS_CLOUD Package Format Options for EXPORT_DATA.

   Il valore priority determina il grado di parallelismo per il recupero dei dati dal database.

   Il valore interval specifica l'intervallo di tempo in minuti tra le esecuzioni consecutive di un job pipeline. L'intervallo predefinito è interval di 15 minuti.

   Per informazioni dettagliate sugli attributi della pipeline, vedere Attributi DBMS_CLOUD_PIPELINE.

   Dopo aver creato una pipeline, è possibile eseguire il test della pipeline o avviarla:

   • Esegui test pipeline

   • Avvia una pipeline

Esegui test pipeline

Utilizzare RUN_PIPELINE_ONCE per eseguire una pipeline una volta su richiesta senza creare un job pianificato.

RUN_PIPELINE_ONCE è utile per eseguire il test di una pipeline prima di avviare la pipeline. Dopo aver eseguito una sola volta una pipeline per eseguire il test della pipeline e verificare che funzioni come previsto, utilizzare RESET_PIPELINE per reimpostare lo stato della pipeline (allo stato precedente all'esecuzione di RUN_PIPELINE_ONCE).

1. creare una pipeline;

   Per ulteriori informazioni, vedere Creare e configurare una pipeline per il caricamento dei dati.

2. Eseguire una sola volta una pipeline per eseguire il test della pipeline.

    BEGIN
        DBMS_CLOUD_PIPELINE.RUN_PIPELINE_ONCE(
            pipeline_name => 'MY_PIPE1'
    );
    END;
    /
   

   Per ulteriori informazioni, vedere RUN_PIPELINE_ONCE Procedure.

3. Eseguire i controlli necessari per verificare che la pipeline funzioni come previsto.

   Per ulteriori informazioni, vedere Monitoraggio e risoluzione dei problemi delle pipeline.

4. Reimpostare la pipeline.

    BEGIN
       DBMS_CLOUD_PIPELINE.RESET_PIPELINE(
         pipeline_name => 'MY_PIPE1',
         purge_data => TRUE
    );
    END;
    /
   

   Per ulteriori informazioni, vedere Procedura RESET_PIPELINE.