Pipelines erstellen und konfigurieren

Sie können eine oder mehrere Lade- oder Exportpipelines erstellen. Wenn Sie eine Pipeline erstellen, verwenden Sie Parameter und legen Pipelineattribute fest, um die Pipeline zu konfigurieren.

Folgende Optionen stehen zum Erstellen und Konfigurieren einer Pipeline zur Verfügung:

Pipeline zum Laden von Daten erstellen und konfigurieren

Sie können eine Pipeline erstellen, um Daten aus externen Dateien im Objektspeicher oder Verzeichnisse in Tabellen in Autonomous AI Database zu laden.

Eine Ladepipeline verbraucht Daten, die im Objektspeicher oder in Verzeichnissen gespeichert sind, und lädt sie in eine Tabelle in Autonome KI-Datenbank. Wenn Sie eine Ladepipeline erstellen, wird die Pipeline in regelmäßigen Abständen ausgeführt, um Daten zu konsumieren, die am Quellspeicherort platziert wurden. Wenn neue Datendateien ankommen, lädt die Pipeline die neuen Daten. Sie können auch eine Pipeline verwenden, um Dateien mit Lebenslauf- und Wiederholungsfunktionen zuverlässig aus dem Quellspeicherort in eine Tabelle in Ihrer Datenbank zu kopieren.

Bei einer Ladepipeline verwendet das Pipelinepackage DBMS_CLOUD.COPY_DATA zum Laden von Daten.

Verwenden Sie in Ihrer autonomen KI-Datenbank entweder eine vorhandene Tabelle, oder erstellen Sie die Datenbanktabelle, in die Sie Daten laden. Beispiel: 

CREATE TABLE EMPLOYEE
            (name     VARCHAR2(128),
             age      NUMBER,
             salary   NUMBER);
  1. Pipeline zum Laden von Daten aus dem Objektspeicher oder aus Directory-Objekten erstellen
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name => 'MY_PIPE1',
        pipeline_type => 'LOAD',
        description   => 'Load metrics from object store into a table'
  );
END;
/

    Weitere Informationen finden Sie unter Prozedur CREATE_PIPELINE.

  2. Erstellen Sie ein Zugangsdatenobjekt, um auf den Objektspeicher zuzugreifen, der die Dateien enthält, die Sie laden.

    Sie geben die Zugangsdaten für den Speicherort der Pipelinequelle mit dem Attribut credential_name an. Wenn Sie im nächsten Schritt keine credential_name angeben, wird der credential_name-Wert auf NULL gesetzt. Sie können den Standardwert NULL verwenden, wenn das Attribut location eine öffentliche oder vorab authentifizierte URL ist.

    Weitere Informationen finden Sie unter Prozedur CREATE_CREDENTIAL.

  3. Legen Sie die Pipelineattribute einschließlich der erforderlichen Attribute fest: location, table_name und format.
    Fall 1: Erstellen Sie eine Pipeline zum Laden von Daten aus dem Objektspeicher.
    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;
/
    Fall 2: Pipeline zum Laden von Daten aus Directory-Objekten erstellen
    
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;
/

    Die folgenden Attribute müssen festgelegt werden, um eine Ladepipeline in beiden Fällen auszuführen:

    • location: Gibt den Speicherort der Quelldatei im Objektspeicher oder Verzeichnisobjekt an.

    • table_name: Gibt die Tabelle in der Datenbank an, in die Sie Daten laden. Die angegebene location gilt für eine table_name pro Pipeline.

    • format: Beschreibt das Format der Daten, die Sie laden.

      Weitere Informationen finden Sie unter DBMS_CLOUD-Package - Formatoptionen.

    credential_name ist die Zugangsdaten, die Sie im vorherigen Schritt erstellt haben.

    Der Wert priority bestimmt die Anzahl der parallel geladenen Dateien. Eine Pipeline mit einer höheren Priorität belegt mehr Datenbankressourcen und schließt jede Ausführung schneller ab als eine Ausführung mit einer niedrigeren Priorität.

    Der Wert interval gibt das Zeitintervall in Minuten zwischen aufeinanderfolgenden Ausführungen eines Pipelinejobs an. interval ist standardmäßig 15 Minuten.

    Details zu den Pipelineattributen finden Sie unter DBMS_CLOUD_PIPELINE-Attribute.

  4. Nachdem Sie eine Pipeline erstellt haben, können Sie die Pipeline testen oder starten:

Alternativ können Sie das folgende Format verwenden, um das Format für JSON festzulegen:

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

Pipeline zum Exportieren mit Zeitstempelspalte erstellen und konfigurieren

Sie können eine Exportpipeline erstellen, um Zeitreihendaten automatisch aus Ihrer autonomen KI-Datenbank in den Objektspeicher zu exportieren.

Mit dieser Exportpipelineoption geben Sie eine Tabelle oder SQL-Abfrage und eine Spalte mit einem Zeitstempel an, mit dem die Pipeline die Zeit des letzten Uploads verfolgt. Mit einer Exportpipeline können Sie Daten für die Nutzung durch andere Anwendungen freigeben oder Daten im Objektspeicher speichern.

Bei einer Exportpipeline verwendet das Pipelinepackage DBMS_CLOUD.EXPORT_DATA zum Exportieren von Daten.

Eine Exportpipeline exportiert Daten aus Ihrer autonomen KI-Datenbank in den Objektspeicher. Wenn Sie eine Exportpipeline erstellen, wird die Pipeline in regelmäßigen Abständen ausgeführt und platziert Daten im Objektspeicher.

  1. Erstellen Sie eine Pipeline zum Exportieren von Daten in den Objektspeicher.
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name=>'EXP_PIPE1',
        pipeline_type=>'EXPORT',
        description=>'Export time series metrics to object store');
END;
/

    Weitere Informationen finden Sie unter Prozedur CREATE_PIPELINE.

  2. Erstellen Sie ein Zugangsdatenobjekt, um auf den Speicherort des Zielobjektspeichers zuzugreifen, in den Sie Datendateien exportieren.

    Sie geben die Zugangsdaten für den Speicherort des Pipelineziels mit dem Attribut credential_name an. Wenn Sie im nächsten Schritt keine credential_name angeben, wird der credential_name-Wert auf NULL gesetzt. Sie können den Standardwert NULL verwenden, wenn das Attribut location eine öffentliche oder vorab authentifizierte URL ist.

    Weitere Informationen finden Sie unter Prozedur CREATE_CREDENTIAL.

  3. Legen Sie die Exportpipelineattribute fest.

    Wenn Sie einen Parameter table_name angeben, werden Tabellenzeilen in den Objektspeicher exportiert. Wenn Sie einen query-Parameter angeben, gibt die Abfrage eine SELECT-Anweisung an, sodass nur die erforderlichen Daten in den Objektspeicher exportiert werden.

    • Verwenden Sie einen table_name-Parameter: 

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

    • Verwenden Sie einen query-Parameter: 

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

    Dabei ist credential_name die Zugangsdaten, die Sie im vorherigen Schritt erstellt haben.

    Die folgenden Attribute müssen festgelegt werden, um eine Exportpipeline auszuführen:

    • location: Gibt den Speicherort des Zielobjektspeichers an. Die angegebene location gilt für eine table_name pro Pipeline.

    • table_name: Gibt die Tabelle in der Datenbank an, die die zu exportierenden Daten enthält (entweder der Parameter table_name oder der Parameter query ist erforderlich).

    • query: Gibt die Abfrage an, die in der Datenbank ausgeführt werden soll und die Daten bereitstellt, die Sie exportieren (entweder der Parameter table_name oder der Parameter query ist erforderlich).

    • format: Beschreibt das Format der Daten, die Sie exportieren.

      Weitere Informationen finden Sie unter DBMS_CLOUD-Packageformatoptionen für EXPORT_DATA.

    Der Wert priority bestimmt den Parallelisierungsgrad für das Abrufen von Daten aus der Datenbank.

    Der Wert interval gibt das Zeitintervall in Minuten zwischen aufeinanderfolgenden Ausführungen eines Pipelinejobs an. interval ist standardmäßig 15 Minuten.

    Details zu den Pipelineattributen finden Sie unter DBMS_CLOUD_PIPELINE-Attribute.

    Nachdem Sie eine Pipeline erstellt haben, können Sie die Pipeline testen oder die Pipeline starten:

Pipeline zum Exportieren von Abfrageergebnissen (ohne Zeitstempel) erstellen und konfigurieren

Sie können eine Exportpipeline erstellen, um Daten automatisch aus Ihrer autonomen KI-Datenbank in den Objektspeicher zu exportieren. Mit dieser Exportpipelineoption geben Sie eine SQL-Abfrage an, die in der Pipeline regelmäßig ausgeführt wird, um Daten in den Objektspeicher zu exportieren. Mit dieser Exportoption können Sie die neuesten Daten aus Ihrer autonomen KI-Datenbank für den Objektspeicher freigeben, damit andere Anwendungen die Daten nutzen können.

Eine Exportpipeline exportiert Daten aus Ihrer autonomen KI-Datenbank in den Objektspeicher. Wenn Sie eine Exportpipeline erstellen, wird die Pipeline in regelmäßigen Abständen ausgeführt und platziert Daten im Objektspeicher.

  1. Erstellen Sie eine Pipeline zum Exportieren von Daten in den Objektspeicher.
    BEGIN
     DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
        pipeline_name=>'EXP_PIPE2',
        pipeline_type=>'EXPORT',
        description=>'Export query results to object store.');
END;
/

    Weitere Informationen finden Sie unter Prozedur CREATE_PIPELINE.

  2. Erstellen Sie ein Zugangsdatenobjekt, um auf den Speicherort des Zielobjektspeichers zuzugreifen, in den Sie Datendateien exportieren.

    Sie geben die Zugangsdaten für den Speicherort des Pipelineziels mit dem Attribut credential_name an. Wenn Sie im nächsten Schritt keine credential_name angeben, wird der credential_name-Wert auf NULL gesetzt. Sie können den Standardwert NULL verwenden, wenn das Attribut location eine öffentliche oder vorab authentifizierte URL ist.

    Weitere Informationen finden Sie unter Prozedur CREATE_CREDENTIAL.

  3. Legen Sie die Exportpipelineattribute fest.
    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;
/

    Dabei ist credential_name die Zugangsdaten, die Sie im vorherigen Schritt erstellt haben.

    Die folgenden Attribute müssen festgelegt werden, um eine Exportpipeline auszuführen:

    • location: Gibt den Speicherort des Zielobjektspeichers an.

    • query: Gibt die Abfrage an, die in der Datenbank ausgeführt werden soll und die Daten bereitstellt, die Sie exportieren.

    • format: Beschreibt das Format der Daten, die Sie exportieren.

      Weitere Informationen finden Sie unter DBMS_CLOUD-Packageformatoptionen für EXPORT_DATA.

    Der Wert priority bestimmt den Parallelisierungsgrad für das Abrufen von Daten aus der Datenbank.

    Der Wert interval gibt das Zeitintervall in Minuten zwischen aufeinanderfolgenden Ausführungen eines Pipelinejobs an. interval ist standardmäßig 15 Minuten.

    Details zu den Pipelineattributen finden Sie unter DBMS_CLOUD_PIPELINE-Attribute.

    Nachdem Sie eine Pipeline erstellt haben, können Sie die Pipeline testen oder die Pipeline starten:

Pipelines testen

Verwenden Sie RUN_PIPELINE_ONCE, um eine Pipeline nach Bedarf auszuführen, ohne einen geplanten Job zu erstellen.

RUN_PIPELINE_ONCE ist nützlich, um eine Pipeline zu testen, bevor Sie die Pipeline starten. Nachdem Sie eine Pipeline einmal ausgeführt haben, um die Pipeline zu testen und zu prüfen, ob sie wie erwartet funktioniert, setzen Sie den Status der Pipeline mit RESET_PIPELINE zurück (auf den Status, bevor Sie RUN_PIPELINE_ONCE ausgeführt haben).

  1. Pipeline erstellen.

    Weitere Informationen finden Sie unter Pipeline zum Laden der Daten erstellen und konfigurieren.

  2. Führen Sie eine Pipeline einmal aus, um die Pipeline zu testen.
    BEGIN
    DBMS_CLOUD_PIPELINE.RUN_PIPELINE_ONCE(
        pipeline_name => 'MY_PIPE1'
);
END;
/

    Weitere Informationen finden Sie unter Prozedur RUN_PIPELINE_ONCE.

  3. Führen Sie alle erforderlichen Prüfungen aus, um zu prüfen, ob die Pipeline wie erwartet funktioniert.

    Weitere Informationen finden Sie unter Pipelines überwachen und Fehler beheben.

  4. Setzen Sie die Pipeline zurück.
    BEGIN  
   DBMS_CLOUD_PIPELINE.RESET_PIPELINE(
     pipeline_name => 'MY_PIPE1',
     purge_data => TRUE
);
END;
/

    Weitere Informationen finden Sie unter Prozedur RESET_PIPELINE.