Utilizzo di Neptune Streams

La funzionalità Neptune Streams consente di generare una sequenza completa di voci di log delle modifiche che registrano automaticamente e in tempo reale ogni modifica apportata ai dati del grafo. Per una panoramica di questa caratteristica, consulta Acquisizione di modifiche al grafo in tempo reale mediante Neptune Streams.

Abilitazione di Neptune Streams

È possibile abilitare o disabilitare Neptune Streams in qualsiasi momento impostando il parametro del cluster database neptune_streams. L'impostazione del parametro su 1 abilita i flussi, mentre l'impostazione su 0 disabilita i flussi.

Nota

Dopo aver modificato il parametro del cluster database neptune_streams, è necessario riavviare tutte le istanze database nel cluster per rendere effettiva la modifica.

È possibile impostare il parametro del cluster database neptune_streams_expiry_days per controllare per quanti giorni, da 1 a 90, i record di flusso rimangono sul server prima di essere eliminati. Il valore predefinito è 7.

Neptune Streams è stato inizialmente introdotto come funzionalità sperimentale che era possibile abilitare o disabilitare in modalità di laboratorio utilizzando il parametro del cluster database neptune_lab_mode (vedi Modalità di laboratorio Neptune). L'utilizzo della modalità laboratorio per abilitare Streams è ora obsoleto e verrà disattivato in futuro.

Disabilitazione di Neptune Streams

Puoi disattivare Neptune Streams in qualsiasi momento in cui è in esecuzione.

Per disattivare Streams, aggiorna il gruppo di parametri Cluster DB in modo che il valore del parametro neptune_streams sia impostato su 0.

Importante

Non appena Streams viene disattivato, non puoi più accedere ai dati di log delle modifiche. Assicurati di leggere ciò che ti interessa prima di disattivare Streams.

Chiamata di REST API di Neptune Streams

Puoi accedere a Neptune Streams utilizzando una REST API che invia una richiesta HTTP GET a uno dei seguenti endpoint locali:

• Per un database grafico SPARQL:   https://Neptune-DNS:8182/sparql/stream.

• Per un database a grafo Gremlin o openCypher: https://Neptune-DNS:8182/propertygraph/stream o https://Neptune-DNS:8182/pg/stream.

Nota

A partire dal rilascio 1.1.0.0 del motore, l'endpoint del flusso Gremlin (https://Neptune-DNS:8182/gremlin/stream) è diventato obsoleto, insieme al formato di output associato (GREMLIN_JSON). È ancora supportato per la compatibilità con le versioni precedenti, ma potrebbe essere rimosso nei rilasci futuri.

È consentita una sola operazione GET HTTP.

Neptune supporta la compressione gzip della risposta, purché la richiesta HTTP includa un'intestazione Accept-Encoding che specifichi gzip come formato di compressione accettato (ovvero "Accept-Encoding: gzip").

Parametri

• limit: long, facoltativo. Intervallo: 1-100.000. Impostazione predefinita: 10

  Specifica il numero massimo di record da restituire. Esiste anche un limite di dimensioni di 10 MB per la risposta che non può essere modificato e che ha la precedenza sul numero di record specificato nel parametro limit. La risposta include un record che supera la soglia se il limite di 10 MB è stato raggiunto.

• iteratorType: stringa, facoltativo.

  Questo parametro può accettare uno dei seguenti valori:

  • AT_SEQUENCE_NUMBER: (valore predefinito) indica che la lettura deve iniziare dal numero di sequenza di eventi specificato congiuntamente dai parametri commitNum e opNum.

  • AFTER_SEQUENCE_NUMBER: indica che la lettura deve iniziare immediatamente dopo il numero di sequenza di eventi specificato congiuntamente dai parametri commitNum e opNum.

  • TRIM_HORIZON: indica che la lettura deve iniziare dall'ultimo record non tagliato nel sistema, ovvero il record più vecchio non scaduto (non ancora eliminato) nel flusso di log delle modifiche. Questa modalità è utile durante l'avvio dell'applicazione, quando non si dispone di un numero di sequenza dell'evento di avvio specifico.

  • LATEST: indica che la lettura deve iniziare dal record più recente non tagliato nel sistema, ovvero il record più recente non scaduto (non ancora eliminato) nel flusso di log delle modifiche. Questa modalità è utile quando è necessario leggere i record dalla parte superiore del flusso per non elaborare i record più vecchi, ad esempio durante un ripristino di emergenza o un aggiornamento senza tempi di inattività. Notare che in questa modalità viene restituito al massimo un solo record.

• commitNum: long, obbligatorio quando iteratorType è AT_SEQUENCE_NUMBER o AFTER_SEQUENCE_NUMBER.

  Il numero di commit del record iniziale da leggere dal flusso change-log.

  Questo parametro viene ignorato quando iteratorType è TRIM_HORIZON o LATEST.

• opNum: long, facoltativo (il valore predefinito è 1).

  Il numero di sequenza dell'operazione all'interno del commit specificato da cui iniziare la lettura nei dati del flusso change-log.

Le operazioni che modificano i dati del grafo SPARQL generano solitamente solo un singolo record di modifica per operazione. Tuttavia, le operazioni che modificano i dati del grafo Gremlin possono generare più record di modifica per operazione, come negli esempi seguenti:

• INSERT: un vertice Gremlin può avere più etichette e un elemento Gremlin può avere più proprietà. Un record di modifica separato viene generato per ogni etichetta e proprietà quando si inserisce un elemento.

• UPDATE: quando la proprietà di un elemento Gremlin viene modificata, vengono generati due record di modifica: il primo per la rimozione del valore precedente e il secondo per l'inserimento del nuovo valore.

• DELETE: viene generato un record di modifica separato per ogni proprietà dell'elemento che viene eliminata. Ad esempio, quando un edge Gremlin con proprietà viene eliminato, viene generato un record di modifica per ciascuna delle proprietà e successivamente ne viene generato uno per l'eliminazione dell'etichetta edge.

  Quando un vertice Gremlin viene eliminato, vengono eliminate per prime tutte le proprietà edge in entrata e in uscita, quindi le etichette edge, le proprietà dei vertici e infine le etichette dei vertici. Ognuna di queste eliminazioni genera un record di modifica.

Formato della risposta API di Neptune Streams

Una risposta a una richiesta REST API di Neptune Streams include i campi seguenti:

• lastEventId: identificatore di sequenza dell'ultima modifica nella risposta del flusso. Un ID evento è composto da due campi: un commitNum che identifica una transazione che ha modificato il grafo e un opNum che identifica un'operazione specifica all'interno di tale transazione. Questo viene mostrato nell'esempio seguente.

    "eventId": {
      "commitNum": 12,
      "opNum": 1
    }

• lastTrxTimestamp: l'ora in cui è stato richiesto il commit per la transazione, in millisecondi dall'epoca Unix.

• format: formato di serializzazione per i record di modifica restituiti. I valori possibili sono PG_JSON per i record di modifica Gremlin o openCypher e NQUADS per i record di modifica SPARQL.

• records: un array di record serializzati del flusso di log delle modifiche inclusi nella risposta. Ogni record dell'array records contiene i seguenti campi:

  • commitTimestamp: l'ora in cui è stato richiesto il commit per la transazione, in millisecondi dall'epoca Unix.

  • eventId: identificatore di sequenza del record di modifica del flusso.

  • data— Il record serializzato di Gremlin, SPARQL o change. OpenCypher I formati di serializzazione di ciascun record sono descritti più dettagliatamente nella sezione successiva, Formati di serializzazione in Neptune Streams.

  • op: operazione che ha creato la modifica.

  • isLastOp: presente solo se questa operazione è l'ultima della transazione. Se presente, è impostato su true. È utile per garantire che venga consumata un'intera transazione.

• totalRecords: numero totale di record nella risposta.

Ad esempio, la seguente risposta restituisce i dati di modifica di Gremlin, per una transazione che contiene più di un'operazione:

{
  "lastEventId": {
    "commitNum": 12,
    "opNum": 1
  },
  "lastTrxTimestamp": 1560011610678,
  "format": "PG_JSON",
  "records": [
    {
      "commitTimestamp": 1560011610678,
      "eventId": {
        "commitNum": 1,
        "opNum": 1
      },
      "data": {
        "id": "d2b59bf8-0d0f-218b-f68b-2aa7b0b1904a",
        "type": "vl",
        "key": "label",
        "value": {
          "value": "vertex",
          "dataType": "String"
        }
      },
      "op": "ADD"
    }
  ],
  "totalRecords": 1
}

La seguente risposta restituisce i dati di modifica SPARQL per l'ultima operazione di una transazione (operazione identificata da EventId(97, 1) nella transazione numero 97).

{
  "lastEventId": {
    "commitNum": 97,
    "opNum": 1
  },
  "lastTrxTimestamp": 1561489355102,
  "format": "NQUADS",
  "records": [
    {
      "commitTimestamp": 1561489355102,
      "eventId": {
        "commitNum": 97,
        "opNum": 1
      },
      "data": {
        "stmt": "<https://test.com/s> <https://test.com/p> <https://test.com/o> .\n"
      },
      "op": "ADD",
      "isLastOp": true
    }
  ],
  "totalRecords": 1
}

Eccezioni API di Neptune Streams

La tabella seguente descrive le eccezioni di Neptune Streams.

Codice di errore	Codice HTTP	OK riprovare?	Messaggio
InvalidParameterException	400	No	È stato fornito un out-of-range valore or non valido come parametro di input.
ExpiredStreamException	400	No	Tutti i record richiesti superano l'età massima consentita e sono scaduti.
ThrottlingException	500	Sì	La velocità delle richieste supera il throughput massimo consentito.
StreamRecordsNotFoundException	404	No	Impossibile trovare la risorsa richiesta. Il flusso potrebbe non essere specificato correttamente.
MemoryLimitExceededException	500	Sì	L'elaborazione della richiesta non è andata a buon fine a causa della mancanza di memoria, tuttavia si potrà riprovare quando il server sarà meno occupato.