Creare e gestire flussi dati

Creare un flusso di dati dall'interfaccia utente

  1. Dal riquadro di navigazione a sinistra, selezionare Flussi di dati per aprire la pagina Flussi di dati.

  2. Fare clic sul pulsante Aggiungi flusso dati. Viene visualizzata la casella Aggiungi flusso di dati diaglox.

    Pagina principale di Data Streams con il pulsante Aggiungi flussi di dati

  3. Nella pagina Informazioni flusso dati immettere un nome risorsa univoco nella casella Nome.

    Il nome deve essere compreso tra 1 e 32 caratteri. Iniziare con una lettera. Utilizzare un nome stabile perché i checkpoint sono definiti nel contratto di flusso. Fare clic su Avanti.

    Finestra di dialogo Aggiungi flussi di dati: pagina Informazioni flusso di dati

  4. Aggiungere la descrizione al proprietario dell'azienda, al trail di origine, alle tabelle, al consumatore e all'obiettivo di recupero.

  5. Nella pagina Opzioni di origine fornire i valori per le opzioni mostrate nell'immagine seguente:

    Finestra di dialogo Flusso di dati: pagina Opzioni di origine

    • Nome percorso: il nome del file trail di origine. Il nome del trail deve essere costituito da due caratteri che identificano il prefisso del trail di origine o l'endpoint di origine accettato dal servizio.

    • Sottodirectory trail: il percorso della sottodirectory in cui vengono memorizzati i file trail. Questa è la directory sotto la distribuzione GoldenGate in cui è memorizzato il trail di origine, quando non si trova nella posizione predefinita.

    • Codifica: questa opzione controlla la codifica dei record per il flusso di dati. Al momento, la codifica JSON è supportata. Utilizzare l'editor YAML o la configurazione REST per le opzioni JSON avanzate.

    • Dimensione buffer: questo valore controlla la dimensione del buffer di memoria utilizzata nel servizio di streaming dati. Il servizio di streaming dati scaricherà la coda dei messaggi in-memory dopo che la dimensione totale in byte supera il valore specificato e consegna i record al client. Intervallo da 131.072 a 10.485.760; valore predefinito 1.048.576. Non si tratta di una conservazione durevole.

    • Quality of Service: questa opzione definisce il funzionamento di eliminazione e ripristino duplicati del flusso di dati. exactlyOnce, atLeastOnce o atMostOnce.

      Fare riferimento alla sezione Semantica checkpoint, recupero e distribuzione prima di selezionare un valore non predefinito.

      Nel servizio di streaming dati sono supportati tre livelli di qualità del servizio:

      • Esattamente una volta: questa modalità è la modalità più restrittiva in cui il servizio filtra i record duplicati durante il recupero della visualizzazione di un record RESTART OK/ABEND negli trail di origine. I client non visualizzeranno record duplicati. Se il servizio non riesce a individuare il record con l'ultima posizione elaborata specificata, si verifica un errore.

      • Almeno una volta: questa modalità non elimina i record duplicati durante il recupero né visualizza i record RESTART OK/ABEND nello storico di origine. I client possono visualizzare record duplicati nel flusso di dati. Inoltre, se il servizio non è in grado di individuare il record con l'ultima posizione elaborata specificata, si verifica un errore.

      • Al massimo una volta: questa modalità elimina i record duplicati durante il recupero o visualizza un record RESTART OK/ABEND nello storico origini. I client non visualizzeranno record duplicati nel flusso di dati. Se il servizio non riesce a individuare il record con l'ultima posizione elaborata specificata, troverà il successivo record disponibile e andrà avanti.

    • Formato CloudEvents: il servizio di streaming dati supporta la trasmissione dei record di dati in formato CloudEvents. Esegue il wrapping dei record GoldenGate in JSON CloudEvents. Disabilitato per impostazione predefinita. Per impostazione predefinita, questo formato è disabilitato e può essere controllato abilitando la proprietà mediante lo switch di attivazione/disattivazione durante la creazione del canale di streaming dati.

  6. Nella pagina Opzioni di filtro sono disponibili le opzioni per includere ed escludere le regole di filtro, come mostrato nell'immagine riportata di seguito.

    Finestra di dialogo Flusso di dati:

    Specificare le opzioni della regola di filtro come indicato di seguito.

    • Azione regola: selezionare le opzioni Escludi o Includi.

    • Tipo di filtro: il tipo di filtro include quanto segue:

      Tipo di oggetto: è possibile selezionare più opzioni dall'elenco a discesa, tra cui DML, DDL, INSERT, UPDATE, UPSERT, DELETE.

      Nomi oggetto: il nome dell'oggetto filtro creato in precedenza.

Fare clic su Aggiungi per aggiungere la regola di filtraggio al processo Flusso di dati.

  1. Fare clic su Crea flusso dati. Verrà visualizzata di nuovo la home page del servizio di distribuzione in cui è elencato il flusso dati.

  2. Fare clic sul flusso di dati creato per visualizzare le definizioni di Data Streaming AsyncAPI del documento YAML per il flusso.

Modifica configurazione flussi di dati dall'interfaccia utente

Per modificare la configurazione dei flussi di dati:

  1. Dal riquadro di navigazione a sinistra, fare clic su Flussi di dati.

  2. Selezionare il nome del flusso di dati da modificare.

  3. Nella pagina Flussi di dati, utilizzare la colonna Azione per visualizzare i dettagli del flusso di dati, eliminare un flusso di dati e modificarne il filtro.

    Modificare il filtro del flusso di dati.

  4. Se si fa clic sull'opzione Modifica filtro nel menu Azione, viene visualizzata la finestra di dialogo Modifica regole filtro flusso dati. Da questa casella è possibile modificare l'azione regola (Includi, Escludi) e il tipo di filtro (nome oggetto o tipo di oggetto).

  5. Fare clic su Aggiungi per applicare il filtro.

  6. Fare clic su Sottometti per restituire la pagina Flussi di dati.

    È possibile modificare ulteriormente il flusso di dati facendo clic sul flusso di dati nella colonna Nome. Viene visualizzata la configurazione completa del flusso dati. Utilizzare l'icona a forma di matita accanto a ciascuna impostazione di configurazione per modificarla. È possibile modificare il file trail di origine utilizzato da un flusso di dati, le relative regole di filtro e la qualità del servizio. È inoltre possibile utilizzare l'editor YAML per modificare la configurazione del flusso di dati e caricare le modifiche utilizzando l'icona Carica modifiche accanto all'editor YAML.

Crea un flusso di dati con l'API REST

Utilizzare POST /services/v2/stream/{streamName} con un account amministratore. La richiesta minima identifica un trail di origine:

{
  "$schema": "ogg:dataStream",
  "source": {
    "trail": "ea",
    "path": "./dirdat"
  }
}

Una richiesta di produzione in genere imposta in modo esplicito la qualità del servizio, i filtri, le emissioni dello schema JSON, i campi di metadati e i controlli delle dimensioni. Il file di supporto data_stream_create_request.json contiene un esempio completo di sanificazione.

{
  "$schema": "ogg:dataStream",
  "description": "Streams sales order changes",
  "source": {"trail": "ea", "path": "./dirdat"},
  "qualityOfService": "exactlyOnce",
  "bufferSize": 1048576,
  "tcpKeepAliveTimeout": 120,
  "cloudEventsFormat": false,
  "rules": [
    {
      "action": "include",
      "filter": {"objectNames": ["SALES.ORDERS", "SALES.ORDER_ITEMS"]}
    }
  ],
  "encoding": {
    "json": {
      "$schema": "ogg:outputFormatJson",
      "model": "op",
      "includeJSONSchemaRecord": true,
      "includeObjectMetadataRecords": true,
      "includeStreamMetadataRecord": true,
      "metaFields": {
        "qualifiedTable": true,
        "opType": true,
        "opTs": true,
        "pos": true,
        "xid": true,
        "csn": "csn",
        "opSeqno": "op_seqno",
        "transInd": "trans_ind"
      }
    }
  }
}

Proprietà di configurazione di livello superiore

La tabella seguente fornisce le proprietà di configurazione utilizzate per configurare il flusso.

Proprietà Tipo/predefinito Descrizione
$schema Stringa/ogg:dataStream Identificativo dello schema di configurazione.
source Oggetto/obbligatorio Percorso di origine e percorso facoltativo.
description Stringa / nessuna Da 1 a 4.096 caratteri che descrivono il flusso di dati.
qualityOfService Enumerazione/exactlyOnce atLeastOnce, exactlyOnce o atMostOnce.
bufferSize Numero intero / 1.048.576 Soglia di flush in byte. Minimo 131.072; massimo 10.485.760.
tcpKeepAliveTimeout Numero intero / 120 Timeout di controllo attività TCP in secondi. Minimo 1; massimo 10.800.
cloudEventsFormat Booleano/false Abilita gli envelope JSON CloudEvents.
rules Array/nessuno Da 1 a 1.024 includono o escludono regole.
encoding Stringa o oggetto / json Utilizzare "json" per le impostazioni predefinite o {"json": {...}} per le impostazioni di output avanzate.

Recupera, aggiorna, elenca ed elimina il flusso di dati

Utilizzare la tabella seguente per gestire o eliminare i flussi di dati:

Task Metodi e percorso Ruolo
Elenca flussi GET /services/v2/stream Utente
Recupera configurazione GET /services/v2/stream/{streamName} Operatore
Recupera YAML generato GET /services/v2/stream/{streamName}/yaml Utente
Aggiorna configurazione PATCH /services/v2/stream/{streamName} Amministratore
Aggiorna YAML PATCH /services/v2/stream/{streamName}/yaml Amministratore
Elimina flusso DELETE /services/v2/stream/{streamName} Amministratore

Attenzione: la modifica dei nomi dei campi, delle stringhe operazione, del modello di record, dei filtri o delle emissioni dello schema modifica il contratto consumer. Coordinare la modifica, eseguire la versione del contratto e riavviare il test da un checkpoint esistente prima del rollout della produzione.

Usa YAML generato come contratto distribuito

L'YAML generato include il server specifico della distribuzione, l'indirizzo del canale, il parametro di query begin, il payload del messaggio e gli schemi di record supportati. Recuperarlo dopo la creazione e dopo ogni modifica alla configurazione. Memorizza una copia revisionata con la release consumer, ma confrontala sempre con la versione runtime durante la convalida della distribuzione.