Criar e Configurar Pipelines

Você pode criar um ou mais pipelines de carga ou exportação. Ao criar um pipeline, você usa parâmetros e define atributos de pipeline para configurar o pipeline.

As opções para criar e configurar um pipeline são as seguintes:

Carregar Pipeline:
- Consulte Criar e Configurar um Pipeline para Carregar Dados.
Exportar Pipeline:
- Exporte resultados incrementais de uma consulta para o armazenamento de objetos usando uma coluna de data ou timestamp como chave para rastrear dados mais recentes. Consulte Criar e Configurar um Pipeline de Exportação com Coluna de Timestamp.
- Exporte dados incrementais de uma tabela para armazenamento de objetos usando uma coluna de data ou timestamp como chave para rastrear dados mais recentes. Consulte Criar e Configurar um Pipeline de Exportação com Coluna de Timestamp.
- Exporte dados de uma tabela para armazenamento de objetos usando uma consulta para selecionar dados sem uma referência a uma coluna de data ou timestamp (para que o pipeline exporte todos os dados que a consulta seleciona para cada execução do scheduler). Consulte Criar e Configurar um Pipeline para Exportar Resultados da Consulta (Sem um Timestamp).

Criar e Configurar um Pipeline para Carregar Dados

Você pode criar um pipeline para carregar dados de arquivos externos no armazenamento de objetos ou em diretórios para tabelas no Autonomous AI Database.

Um pipeline de carregamento consome dados colocados no armazenamento de objetos ou em diretórios e os carrega em uma tabela no Autonomous AI Database. Quando você cria um pipeline de carga, ele é executado em intervalos regulares para consumir dados colocados no local de origem, quando novos arquivos de dados chegam, o pipeline carrega os novos dados. Você também pode usar um pipeline para copiar arquivos de forma confiável, com recursos de retomada e repetição, do local de origem para uma tabela no seu banco de dados.

Com um pipeline de carregamento, o pacote de pipeline usa DBMS_CLOUD.COPY_DATA para carregar dados.

No Autonomous AI Database, use uma tabela existente ou crie a tabela de banco de dados na qual você está carregando dados. Por exemplo:

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

Crie um pipeline para carregar dados do armazenamento de objetos ou de objetos de diretório.

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

Consulte Procedimento CREATE_PIPELINE para obter mais informações.

Crie um objeto de credencial para acessar o armazenamento de objetos que contém os arquivos que você está carregando.

Especifique a credencial para o local de origem do pipeline com o atributo credential_name. Se você não fornecer uma credential_name na próxima etapa, o valor credential_name será definido como NULL. Você pode usar o valor NULL padrão quando o atributo location for um URL público ou pré-autenticado.

Consulte Procedimento CREATE_CREDENTIAL para obter mais informações.
Defina os atributos do pipeline, incluindo os atributos necessários: location, table_name e format.

Caso 1: Criar um pipeline para carregar dados do armazenamento de objetos.
```
 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: Criar um pipeline para carregar dados de objetos de diretório.
```
 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;
 /
```
Os seguintes atributos devem ser definidos para executar um pipeline de carga em ambos os casos:
- location: Especifica o local do arquivo de origem no armazenamento de objetos ou no objeto de diretório.
- table_name: Especifica a tabela no banco de dados em que você está carregando dados. O location especificado é para um table_name por pipeline.
- format: descreve o formato dos dados que você está carregando.
  
  Consulte Opções de Formato de Pacote DBMS_CLOUD para obter mais informações.
A credential_name é a credencial criada na etapa anterior.

O valor priority determina o número de arquivos carregados em paralelo. Um pipeline com prioridade mais alta consome mais recursos de banco de dados e conclui cada execução mais rapidamente, em comparação com a execução em prioridade mais baixa.

O valor interval especifica o intervalo de tempo em minutos entre execuções consecutivas de um job de pipeline. O padrão interval é 15 minutos.

Consulte Atributos DBMS_CLOUD_PIPELINE para obter detalhes sobre os atributos do pipeline.
Depois de criar um pipeline, você poderá testar o pipeline ou iniciar o pipeline:
- Testar Pipelines
- Iniciar um Pipeline

Como alternativa, para definir o formato para JSON, você pode usar o seguinte 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;
/

Criar e Configurar um Pipeline para a Coluna Exportar com Timestamp

Você pode criar um pipeline de exportação para exportar automaticamente dados de série temporal do Autonomous AI Database para o armazenamento de objetos.

Usando essa opção de pipeline de exportação, você especifica uma consulta de tabela ou SQL e uma coluna com um timestamp que o pipeline usa para controlar o horário do último upload. Você pode usar um pipeline de exportação para compartilhar dados para consumo por outros aplicativos ou para salvar dados no armazenamento de objetos.

Com um pipeline de exportação, o pacote de pipeline usa DBMS_CLOUD.EXPORT_DATA para exportar dados.

Um pipeline de exportação exporta dados do seu Autonomous AI Database para o armazenamento de objetos. Quando você cria um pipeline de exportação, o pipeline é executado em intervalos regulares e coloca dados no armazenamento de objetos.

Crie um pipeline para exportar dados para o armazenamento de objetos.

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

Consulte Procedimento CREATE_PIPELINE para obter mais informações.

Crie um objeto de credencial para acessar o local do armazenamento de objetos de destino no qual você está exportando arquivos de dados.

Especifique a credencial para o local de destino do pipeline com o atributo credential_name. Se você não fornecer uma credential_name na próxima etapa, o valor credential_name será definido como NULL. Você pode usar o valor NULL padrão quando o atributo location for um URL público ou pré-autenticado.

Consulte Procedimento CREATE_CREDENTIAL para obter mais informações.
Defina os atributos do pipeline de exportação.

Quando você especifica um parâmetro table_name, as linhas da tabela são exportadas para o armazenamento de objetos. Quando você especifica um parâmetro query, a consulta especifica uma instrução SELECT para que somente os dados necessários sejam exportados para o armazenamento de objetos.
- Usando um parâmetro 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;
/
```
- Usando um parâmetro 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;
/
```
  Onde credential_name é a credencial criada na etapa anterior.
Os seguintes atributos devem ser definidos para executar um pipeline de exportação:
- location: Especifica o local do armazenamento de objetos de destino. O location especificado é para um table_name por pipeline.
- table_name: Especifica a tabela no banco de dados que contém os dados que você está exportando (o parâmetro table_name ou o parâmetro query é obrigatório).
- query: Especifica a consulta a ser executada no banco de dados que fornece os dados que você está exportando (o parâmetro table_name ou o parâmetro query é obrigatório).
- format: Descreve o formato dos dados que você está exportando.
  
  Consulte Opções de Formato de Pacote DBMS_CLOUD para EXPORT_DATA para obter mais informações.
O valor priority determina o grau de paralelismo para extrair dados do banco de dados.

O valor interval especifica o intervalo de tempo em minutos entre execuções consecutivas de um job de pipeline. O padrão interval é 15 minutos.

Consulte Atributos DBMS_CLOUD_PIPELINE para obter detalhes sobre os atributos do pipeline.

Depois de criar um pipeline, você poderá testar o pipeline ou iniciar o pipeline:
- Testar Pipelines
- Iniciar um Pipeline

Criar e Configurar um Pipeline para Exportar Resultados de Consulta (sem um Timestamp)

Você pode criar um pipeline de exportação para exportar automaticamente dados do Autonomous AI Database para o armazenamento de objetos. Usando essa opção de pipeline de exportação, você especifica uma consulta SQL que o pipeline executa periodicamente para exportar dados para o armazenamento de objetos. Você pode usar essa opção de exportação para compartilhar os dados mais recentes do Autonomous AI Database para o armazenamento de objetos de outros aplicativos para consumir os dados.

Crie um pipeline para exportar dados para o armazenamento de objetos.

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

Consulte Procedimento CREATE_PIPELINE para obter mais informações.

Crie um objeto de credencial para acessar o local do armazenamento de objetos de destino no qual você está exportando arquivos de dados.

Especifique a credencial para o local de destino do pipeline com o atributo credential_name. Se você não fornecer uma credential_name na próxima etapa, o valor credential_name será definido como NULL. Você pode usar o valor NULL padrão quando o atributo location for um URL público ou pré-autenticado.

Consulte Procedimento CREATE_CREDENTIAL para obter mais informações.
Defina os atributos do pipeline de exportação.
```
 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;
 /
```
Onde credential_name é a credencial criada na etapa anterior.

Os seguintes atributos devem ser definidos para executar um pipeline de exportação:
- location: Especifica o local do armazenamento de objetos de destino.
- query: Especifica a consulta a ser executada no banco de dados que fornece os dados que você está exportando.
- format: Descreve o formato dos dados que você está exportando.
  
  Consulte Opções de Formato de Pacote DBMS_CLOUD para EXPORT_DATA para obter mais informações.
O valor priority determina o grau de paralelismo para extrair dados do banco de dados.

O valor interval especifica o intervalo de tempo em minutos entre execuções consecutivas de um job de pipeline. O padrão interval é 15 minutos.

Consulte Atributos DBMS_CLOUD_PIPELINE para obter detalhes sobre os atributos do pipeline.

Depois de criar um pipeline, você poderá testar o pipeline ou iniciar o pipeline:
- Testar Pipelines
- Iniciar um Pipeline

Testar Pipelines

Use RUN_PIPELINE_ONCE para executar um pipeline uma vez sob demanda sem criar um job programado.

O RUN_PIPELINE_ONCE é útil para testar um pipeline antes de iniciar o pipeline. Depois de executar um pipeline uma vez para testar o pipeline e verificar se ele está funcionando conforme esperado, use RESET_PIPELINE para redefinir o estado do pipeline (para o estado antes de executar RUN_PIPELINE_ONCE).

Crie um pipeline.

Consulte Criar e Configurar um Pipeline para Carregar Dados para obter mais informações.

Execute um pipeline uma vez para testar o pipeline.

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

Consulte Procedimento RUN_PIPELINE_ONCE para obter mais informações.

Execute as verificações necessárias para verificar se o pipeline está funcionando conforme o esperado.

Consulte Monitorar e Solucionar Problemas de Pipelines para obter mais informações.

Redefina o pipeline.

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

Consulte Procedimento RESET_PIPELINE para obter mais informações.