Oracle Cloud Infrastructureドキュメント

パイプラインの作成および構成

1つ以上のロード・パイプラインまたはエクスポート・パイプラインを作成できます。パイプラインを作成する場合は、パラメータを使用し、パイプライン属性を設定してパイプラインを構成します。

パイプラインを作成および構成するオプションは次のとおりです:

データをロードするためのパイプラインの作成および構成

パイプラインを作成して、オブジェクト・ストア内の外部ファイルまたはディレクトリからAutonomous AI Database内の表にデータをロードできます。

ロード・パイプラインは、オブジェクト・ストアまたはディレクトリに配置されたデータを消費し、Autonomous AI Databaseの表にロードします。ロード・パイプラインを作成すると、パイプラインは定期的に実行され、ソースの場所に配置されたデータが消費されます。新しいデータファイルが到着すると、パイプラインは新しいデータをロードします。パイプラインを使用して、再開および再試行機能を使用して、ソースの場所からデータベース上の表にファイルを確実にコピーすることもできます。

ロード・パイプラインでは、パイプライン・パッケージはDBMS_CLOUD.COPY_DATAを使用してデータをロードします。

自律型AIデータベースで、既存の表を使用するか、データをロードするデータベース表を作成します。たとえば:

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

  1. オブジェクト・ストアまたはディレクトリ・オブジェクトからデータをロードするパイプラインを作成します。

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

    詳細は、CREATE_PIPELINEプロシージャを参照してください。

  2. ロードするファイルを含むオブジェクト・ストアにアクセスするための資格証明オブジェクトを作成します。

    属性credential_nameを使用して、パイプライン・ソースの場所の資格証明を指定します。次のステップでcredential_nameを指定しない場合、credential_name値はNULLに設定されます。location属性がパブリックURLまたは事前認証済URLの場合、デフォルトのNULL値を使用できます。

    詳細は、「CREATE_CREDENTIALプロシージャ」を参照してください。

  3. 必要な属性を含むパイプライン属性(location、table_nameおよびformat)を設定します。

    ケース1: オブジェクト・ストアからデータをロードするためのパイプラインを作成します。

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

    ケース2: ディレクトリ・オブジェクトからデータをロードするためのパイプラインを作成します。

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

    ケース3: ワイルドカードを使用してAmazon S3からCSVファイルをロードするためのパイプラインを作成します。

    ソース・ファイルがAmazon S3に格納されている場合は、AWS資格証明とともにAmazon S3 URLを使用します。この例では、オブジェクト・ストアの場所にワイルドカードを使用して、パターンに一致するファイルのみをロードします。

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

    ケース4: OCI Object StorageからParquetファイルをロードするためのパイプラインを作成します。

    ソース・ファイルがOCIバケットに格納されている場合は、OCI資格証明とともにOCIオブジェクト・ストレージURLを使用します。この例では、バケット接頭辞からParquetファイルをロードします。

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

    ケース5: オブジェクト・ストアからOracle Data Pumpファイルをロードするためのパイプラインを作成します。

    ソース・ファイルがOracle Data Pumpダンプ・ファイルである場合は、datapump形式を使用します。この例では、OCIオブジェクト・ストレージからダンプ・ファイルをロードします。

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

    ケース6: Azure Blob StorageからCSVファイルをロードするためのパイプラインを作成します。

    ソース・ファイルがAzure Blob Storageに格納されている場合は、Azureサービス・プリンシパル資格証明とともにAzure Blob Storage URLを使用します。

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

    各ケースでロード・パイプラインを実行するには、次の属性を設定する必要があります。

    credential_nameは、前のステップで作成した資格証明です。

    priority値は、パラレルにロードされるファイルの数を決定します。優先度の高いパイプラインは、より低い優先度での実行と比較して、より多くのデータベース・リソースを消費し、各実行を高速に完了します。

    interval値は、パイプライン・ジョブの連続実行間の時間間隔を分単位で指定します。デフォルトのintervalは15分です。

    パイプライン属性の詳細は、「DBMS_CLOUD_PIPELINEの属性」を参照してください。

  4. パイプラインを作成した後、パイプラインをテストするか、パイプラインを開始できます:

かわりに、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;
/

タイムスタンプ列を使用したエクスポートのためのパイプラインの作成および構成

エクスポート・パイプラインを作成して、Autonomous AI Databaseからオブジェクト・ストアに時系列データを自動的にエクスポートできます。

このエクスポート・パイプライン・オプションを使用して、表またはSQL問合せと、パイプラインが最終アップロードの時間を追跡するために使用するタイムスタンプを持つ列を指定します。エクスポート・パイプラインを使用して、他のアプリケーションによる消費のためにデータを共有したり、データをオブジェクト・ストアに保存したりできます。

エクスポート・パイプラインでは、パイプライン・パッケージはDBMS_CLOUD.EXPORT_DATAを使用してデータをエクスポートします。

エクスポート・パイプラインは、Autonomous AI Databaseからオブジェクト・ストアにデータをエクスポートします。エクスポート・パイプラインを作成すると、パイプラインは定期的に実行され、データがオブジェクト・ストアに配置されます。

  1. オブジェクト・ストアにデータをエクスポートするパイプラインを作成します。

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

    詳細は、CREATE_PIPELINEプロシージャを参照してください。

  2. 資格証明オブジェクトを作成して、データファイルをエクスポートする宛先オブジェクト・ストアの場所にアクセスします。

    パイプライン宛先の場所の資格証明は、属性credential_nameを使用して指定します。次のステップでcredential_nameを指定しない場合、credential_name値はNULLに設定されます。location属性がパブリックURLまたは事前認証済URLの場合、デフォルトのNULL値を使用できます。

    詳細は、「CREATE_CREDENTIALプロシージャ」を参照してください。

  3. エクスポート・パイプライン属性を設定します。

    table_nameパラメータを指定すると、表の行がオブジェクト・ストアにエクスポートされます。queryパラメータを指定すると、必要なデータのみがオブジェクト・ストアにエクスポートされるように、問合せによってSELECT文が指定されます。

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

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

    問合せ結果をCSVファイルとしてOCIオブジェクト・ストレージにエクスポートし、ヘッダー行に列名を指定します:

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

    表データをParquetファイルとして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;
/

    問合せ結果をOracle Data Pumpファイルとしてオブジェクト・ストアにエクスポートします。

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

    問合せ結果をJSONファイルとしてAzure Blob Storageにエクスポートします:

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

    ここで、credential_nameは、前のステップで作成した資格証明です。

    エクスポート・パイプラインを実行するには、次の属性を設定する必要があります。

    • location: 宛先オブジェクト・ストアの場所を指定します。指定するlocationは、パイプラインごとに1つのtable_name用です。

    • table_name: エクスポートするデータを含むデータベース内の表を指定します。table_nameパラメータまたはqueryパラメータのいずれかを指定します。

    • query: エクスポートするデータを提供するデータベースで実行する問合せを指定します。table_nameパラメータまたはqueryパラメータのいずれかを指定します。

    • format: エクスポートするデータの形式を示します。

      詳細は、「EXPORT_DATAのDBMS_CLOUDパッケージ形式オプション」を参照してください。

      エクスポート・パイプラインの場合、フォーマット・タイプは、DBMS_CLOUD.EXPORT_DATAでサポートされている出力ファイル・タイプ(CSV、JSON、Parquet、XMLおよびOracle Data Pump)に従います。

    priority値は、データベースからデータをフェッチするための並列度を決定します。

    interval値は、パイプライン・ジョブの連続実行間の時間間隔を分単位で指定します。デフォルトのintervalは15分です。

    パイプライン属性の詳細は、「DBMS_CLOUD_PIPELINEの属性」を参照してください。

    パイプラインを作成した後、パイプラインをテストするか、パイプラインを開始できます:

問合せ結果をエクスポートするためのパイプラインの作成および構成(タイムスタンプなし)

エクスポート・パイプラインを作成して、Autonomous AI Databaseからオブジェクト・ストアにデータを自動的にエクスポートできます。このエクスポート・パイプライン・オプションを使用して、パイプラインが定期的に実行し、データをオブジェクト・ストアにエクスポートするSQL問合せを指定します。このエクスポート・オプションを使用して、Autonomous AI Databaseの最新データを、他のアプリケーションがデータを消費するためのオブジェクト・ストアに共有できます。

エクスポート・パイプラインは、Autonomous AI Databaseからオブジェクト・ストアにデータをエクスポートします。エクスポート・パイプラインを作成すると、パイプラインは定期的に実行され、データがオブジェクト・ストアに配置されます。

  1. オブジェクト・ストアにデータをエクスポートするパイプラインを作成します。

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

    詳細は、CREATE_PIPELINEプロシージャを参照してください。

  2. 資格証明オブジェクトを作成して、データファイルをエクスポートする宛先オブジェクト・ストアの場所にアクセスします。

    パイプライン宛先の場所の資格証明は、属性credential_nameを使用して指定します。次のステップでcredential_nameを指定しない場合、credential_name値はNULLに設定されます。location属性がパブリックURLまたは事前認証済URLの場合、デフォルトのNULL値を使用できます。

    詳細は、「CREATE_CREDENTIALプロシージャ」を参照してください。

  3. エクスポート・パイプライン属性を設定します。

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

    ここで、credential_nameは、前のステップで作成した資格証明です。

    エクスポート・パイプラインを実行するには、次の属性を設定する必要があります。

    • location: 宛先オブジェクト・ストアの場所を指定します。

    • query: エクスポートするデータを提供するデータベースで実行する問合せを指定します。

    • format: エクスポートするデータの形式を示します。

      詳細は、「EXPORT_DATAのDBMS_CLOUDパッケージ形式オプション」を参照してください。

    priority値は、データベースからデータをフェッチするための並列度を決定します。

    interval値は、パイプライン・ジョブの連続実行間の時間間隔を分単位で指定します。デフォルトのintervalは15分です。

    パイプライン属性の詳細は、「DBMS_CLOUD_PIPELINEの属性」を参照してください。

    パイプラインを作成した後、パイプラインをテストするか、パイプラインを開始できます: