Buffering persistente Divisione Concatenamento Code DLQ Gestione degli indici End-to-end ) Riconoscimento Contropressione alla fonte

Panoramica delle funzionalità della pipeline in Amazon Ingestion OpenSearch

Amazon OpenSearch Ingestion fornisce pipeline, che consistono in una fonte, un buffer, zero o più processori e uno o più sink. Le pipeline di ingestione sono alimentate da Data Prepper come motore di dati. Per una panoramica dei vari componenti di una pipeline, vedere. Concetti chiave

Le seguenti sezioni forniscono una panoramica di alcune delle funzionalità più comunemente utilizzate in Amazon OpenSearch Ingestion.

Nota

Questo non è un elenco esaustivo delle funzionalità disponibili per le pipeline. Per una documentazione completa di tutte le funzionalità della pipeline disponibili, consulta la documentazione di Data Prepper. Tieni presente che OpenSearch Ingestion impone alcuni vincoli ai plugin e alle opzioni che puoi utilizzare. Per ulteriori informazioni, consulta Plugin e opzioni supportati per le pipeline di Amazon OpenSearch Ingestion.

Buffering persistente

Un buffer persistente archivia i dati in un buffer basato su disco su più zone di disponibilità per aggiungere durabilità ai dati. È possibile utilizzare il buffering persistente per importare dati da tutte le fonti basate su push supportate senza la necessità di configurare un buffer autonomo. Questi includono HTTP e OpenTelemetry sorgenti per log, tracce e metriche.

Per abilitare il buffering persistente, scegli Abilita buffer persistente durante la creazione o l'aggiornamento di una pipeline. Per ulteriori informazioni, consulta. Creazione di pipeline Amazon OpenSearch Ingestion OpenSearch L'ingestione determina automaticamente la capacità di buffering richiesta in base alle Ingestion OpenSearch Compute Unit (Ingestion OCU) specificate per la pipeline.

Per impostazione predefinita, le pipeline utilizzano an per crittografare i dati del buffer. Chiave di proprietà di AWS Queste pipeline non necessitano di autorizzazioni aggiuntive per il ruolo della pipeline. In alternativa, puoi specificare una chiave gestita dal cliente e aggiungere le seguenti autorizzazioni IAM al ruolo della pipeline:


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KeyAccess",
            "Effect": "Allow",
            "Action": [
              "kms:Decrypt",
              "kms:GenerateDataKeyWithoutPlaintext"
            ],
            "Resource": "arn:aws:kms:{region}:{aws-account-id}:key/1234abcd-12ab-34cd-56ef-1234567890ab"
        }
    ]
}

Per ulteriori informazioni, consulta Customer managed keys nella Guida per sviluppatori AWS Key Management Service .

Nota

Se disabiliti il buffering persistente, la pipeline verrà aggiornata per essere eseguita interamente con il buffering in memoria.

Ottimizzazione della dimensione massima del payload della richiesta

Se abiliti il buffering persistente per una pipeline, la dimensione massima del payload della richiesta è predefinita a 1 MB. Il valore predefinito offre le migliori prestazioni. Tuttavia, puoi aumentare questo valore se i tuoi clienti inviano richieste superiori a 1 MB. Per ottimizzare la dimensione massima del payload, impostate l'max_request_lengthopzione all'interno della configurazione di origine. Proprio come il buffering persistente, questa opzione è supportata solo per HTTP e per OpenTelemetry le fonti per log, tracce e metriche.

Gli unici valori validi per l'max_request_lengthopzione sono 1 MB, 1,5 MB, 2 MB, 2,5 MB, 3 MB, 3,5 MB e 4 MB. Se si specifica un valore diverso, viene visualizzato un errore.

L'esempio seguente mostra come configurare la dimensione massima del payload all'interno di una configurazione di pipeline:


...
log-pipeline:
  source:
    http:
      path: "/${pipelineName}/logs"
      max_request_length: 4mb
  processor:
...

Se non abilitate il buffering persistente per una pipeline, il valore predefinito dell'max_request_lengthopzione è 10 MB per tutte le fonti e non può essere modificato.

Divisione

È possibile configurare una pipeline di OpenSearch ingestione per suddividere gli eventi in entrata in una sottopipeline, in modo da eseguire diversi tipi di elaborazione sullo stesso evento in entrata.

La pipeline di esempio seguente divide gli eventi in entrata in due sotto-pipeline. Ogni sottopipeline utilizza il proprio processore per arricchire e manipolare i dati, quindi invia i dati a indici diversi. OpenSearch


version: "2"
log-pipeline:
  source:
    http:
    ...
  sink:
    - pipeline:
        name: "logs_enriched_one_pipeline"
    - pipeline:
        name: "logs_enriched_two_pipeline"

logs_enriched_one_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
        index: "enriched_one_logs"

logs_enriched_two_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_two_logs"

Concatenamento

È possibile concatenare più sotto-pipeline per eseguire l'elaborazione e l'arricchimento dei dati in blocchi. In altre parole, è possibile arricchire un evento in entrata con determinate funzionalità di elaborazione in una sottopipeline, quindi inviarlo a un'altra sottopipeline per un ulteriore arricchimento con un processore diverso e infine inviarlo al relativo sink. OpenSearch

Nell'esempio seguente, la log_pipeline sub-pipeline arricchisce un evento di registro in entrata con un set di processori, quindi invia l'evento a un indice denominato. OpenSearch enriched_logs La pipeline invia lo stesso evento alla log_advanced_pipeline pipeline secondaria, che lo elabora e lo invia a un indice diverso denominato. OpenSearch enriched_advanced_logs


version: "2"
log-pipeline:
  source:
    http:
    ...
  processor:
    ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_logs"
    - pipeline:
        name: "log_advanced_pipeline"

log_advanced_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_advanced_logs"

Code DLQ

Le code DLQ (Dead-letter Queues) sono destinazioni per eventi che una pipeline non riesce a scrivere in un sink. In OpenSearch Ingestion, è necessario specificare un bucket Amazon S3 con autorizzazioni di scrittura appropriate da utilizzare come DLQ. Puoi aggiungere una configurazione DLQ a ogni sink all'interno di una pipeline. Quando una pipeline incontra errori di scrittura, crea oggetti DLQ nel bucket S3 configurato. Gli oggetti DLQ esistono all'interno di un file JSON come una serie di eventi falliti.

Una pipeline scrive eventi nel DLQ quando viene soddisfatta una delle seguenti condizioni:

I max_retries rubinetti per il OpenSearch lavandino sono esauriti. OpenSearch L'ingestione richiede un minimo di 16 per questa opzione.
Gli eventi vengono rifiutati dal sink a causa di una condizione di errore.

Configurazione

Per configurare una coda di lettere non scritte per una pipeline secondaria, specifica l'dlqopzione all'interno della configurazione sink: opensearch


apache-log-pipeline:
  ...
  sink:
    opensearch:
      dlq:
        s3:
          bucket: "my-dlq-bucket"
          key_path_prefix: "dlq-files"
          region: "us-west-2"
          sts_role_arn: "arn:aws:iam::123456789012:role/dlq-role"

I file scritti su questo S3 DLQ avranno il seguente schema di denominazione:


dlq-v${version}-${pipelineName}-${pluginId}-${timestampIso8601}-${uniqueId}

Per ulteriori informazioni, consulta Dead-Letter Queues (DLQ).

Per istruzioni su come configurare il ruolo, vedere. sts_role_arn Scrittura in una coda di lettere non scritte

Esempio

Considerate il seguente file DLQ di esempio:


dlq-v2-apache-log-pipeline-opensearch-2023-04-05T15:26:19.152938Z-e7eb675a-f558-4048-8566-dac15a4f8343

Ecco un esempio di dati che non sono stati scritti nel sink e che vengono inviati al bucket DLQ S3 per ulteriori analisi:


Record_0	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index		  "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log		    "sample log"
timestamp	    "2023-04-14T10:36:01.070Z"

Record_1	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index               "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log                 "another sample log"
timestamp           "2023-04-14T10:36:01.071Z"

Gestione degli indici

Amazon OpenSearch Ingestion offre molte funzionalità di gestione degli indici, tra cui le seguenti.

Creazione di indici

È possibile specificare un nome di indice in un sink di pipeline e OpenSearch Ingestion crea l'indice quando effettua il provisioning della pipeline. Se esiste già un indice, la pipeline lo utilizza per indicizzare gli eventi in entrata. Se si arresta e si riavvia una pipeline o se si aggiorna la configurazione YAML, la pipeline tenta di creare nuovi indici se non esistono già. Una pipeline non può mai eliminare un indice.

I sinks di esempio seguenti creano due indici quando viene eseguito il provisioning della pipeline:


sink:
  - opensearch:
      index: apache_logs
  - opensearch:
      index: nginx_logs

Generazione di nomi e modelli di indici

È possibile generare nomi di indice dinamici utilizzando variabili dai campi degli eventi in arrivo. Nella configurazione sink, usa il formato string${} per segnalare l'interpolazione delle stringhe e usa un puntatore JSON per estrarre i campi dagli eventi. Le opzioni per sono o. index_type custom management_disabled Poiché l'index_typeimpostazione predefinita è per i OpenSearch domini e custom management_disabled per le raccolte OpenSearch Serverless, può essere lasciata non impostata.

Ad esempio, la seguente pipeline seleziona il metadataType campo dagli eventi in entrata per generare i nomi degli indici.


pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}"

La seguente configurazione continua a generare un nuovo indice ogni giorno o ogni ora.


pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd}"

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd.HH}"

Il nome dell'indice può anche essere una stringa semplice con un modello data-ora come suffisso, ad esempio. my-index-%{yyyy.MM.dd} Quando il sink invia dati a OpenSearch, sostituisce il modello data-ora con l'ora UTC e crea un nuovo indice per ogni giorno, ad esempio. my-index-2022.01.25 Per ulteriori informazioni, consultate la classe. DateTimeFormatter

Questo nome di indice può anche essere una stringa formattata (con o senza un suffisso del modello data-ora), ad esempio. my-${index}-name Quando il sink invia dati a OpenSearch, sostituisce la "${index}" parte con il valore dell'evento in fase di elaborazione. Se il formato è"${index1/index2/index3}", sostituisce il campo index1/index2/index3 con il relativo valore nell'evento.

Generazione degli ID dei documenti

Una pipeline può generare un ID di documento durante l'indicizzazione dei documenti su. OpenSearch Può dedurre questi ID di documento dai campi all'interno degli eventi in arrivo.

Questo esempio utilizza il uuid campo di un evento in entrata per generare un ID di documento.


pipeline:
  ...
  sink:
    opensearch:
      index_type: custom
      index: "metadata-${metadataType}-%{yyyy.MM.dd}" 
      document_id_field: "uuid"

Nell'esempio seguente, il processore Add entries unisce i campi uuid e other_field l'evento in entrata per generare un ID del documento.

L'createazione garantisce che i documenti con ID identici non vengano sovrascritti. La pipeline elimina i documenti duplicati senza alcun tentativo o evento DLQ. Si tratta di un'aspettativa ragionevole per gli autori della pipeline che utilizzano questa azione, poiché l'obiettivo è evitare l'aggiornamento dei documenti esistenti.


pipeline:
  ...
  processor:
   - add_entries:
      entries:
        - key: "my_doc_id_field"
          format: "${uuid}-${other_field}"
  sink:
    - opensearch:
       ...
       action: "create"
       document_id_field: "my_doc_id_field"

Potresti voler impostare l'ID del documento di un evento su un campo di un oggetto secondario. Nell'esempio seguente, il plugin OpenSearch sink utilizza l'oggetto secondario info/id per generare un ID del documento.


sink:
  - opensearch:
       ...
       document_id_field: info/id

Dato il seguente evento, la pipeline genererà un documento con il _id campo impostato su: json001


{
   "fieldA":"arbitrary value",
   "info":{
      "id":"json001",
      "fieldA":"xyz",
      "fieldB":"def"
   }
}

Generazione di ID di routing

È possibile utilizzare l'routing_fieldopzione all'interno del plug-in OpenSearch sink per impostare il valore di una proprietà di routing del documento (_routing) su un valore proveniente da un evento in arrivo.

Il routing supporta la sintassi del puntatore JSON, quindi sono disponibili anche campi annidati, non solo campi di primo livello.


sink:
  - opensearch:
       ...
       routing_field: metadata/id
       document_id_field: id

Dato il seguente evento, il plugin genera un documento con il campo impostato su: _routing abcd


{
   "id":"123",
   "metadata":{
      "id":"abcd",
      "fieldA":"valueA"
   },
   "fieldB":"valueB"
}

Per istruzioni su come creare modelli di indice che le pipeline possono utilizzare durante la creazione dell'indice, vedete Modelli di indice.

End-to-end ) Riconoscimento

OpenSearch Ingestion garantisce la durata e l'affidabilità dei dati tracciandone la trasmissione dalla fonte ai pozzi nelle condutture stateless mediante riconoscimento. end-to-end Attualmente, solo il plug-in sorgente S3 supporta il riconoscimento. end-to-end

Con il end-to-end riconoscimento, il plug-in pipeline source crea un set di riconoscimenti per monitorare un batch di eventi. Riceve un riconoscimento positivo quando tali eventi vengono inviati con successo ai rispettivi sink o un riconoscimento negativo quando nessuno degli eventi non può essere inviato ai rispettivi sink.

In caso di guasto o arresto anomalo di un componente della pipeline, o se una fonte non riceve una conferma, la fonte scade e intraprende le azioni necessarie, come riprovare o registrare l'errore. Se nella pipeline sono configurati più sink o più subpipeline, i riconoscimenti a livello di evento vengono inviati solo dopo che l'evento è stato inviato a tutti i sink di tutte le pipeline secondarie. Se un sink ha un DLQ configurato, i riconoscimenti tengono traccia anche degli eventi scritti nel DLQ. end-to-end

Per abilitare la end-to-end conferma, includi l'opzione nella acknowledgments configurazione di origine:


s3-pipeline:
  source:
    s3:
      acknowledgments: true
...

Contropressione alla fonte

Una pipeline può subire una contropressione quando è impegnata nell'elaborazione dei dati o se i suoi sink sono temporaneamente inattivi o rallentano l'acquisizione dei dati. OpenSearch Ingestion ha diversi modi di gestire la contropressione a seconda del plug-in di origine utilizzato da una pipeline.

Origine HTTP

Le pipeline che utilizzano il plug-in di origine HTTP gestiscono la contropressione in modo diverso a seconda del componente della pipeline che è congestionato:

Buffer: quando i buffer sono pieni, la pipeline inizia a restituire lo stato HTTP REQUEST_TIMEOUT con il codice di errore 408 all'endpoint di origine. Quando i buffer vengono liberati, la pipeline riavvia l'elaborazione degli eventi HTTP.
Thread di origine: quando tutti i thread di origine HTTP sono impegnati nell'esecuzione di richieste e la dimensione della coda delle richieste non elaborate ha superato il numero massimo consentito di richieste, la pipeline inizia a restituire lo stato HTTP TOO_MANY_REQUESTS con il codice di errore 429 all'endpoint di origine. Quando la coda delle richieste scende al di sotto della dimensione massima consentita, la pipeline riavvia l'elaborazione delle richieste.

Fonte Otel

Quando i buffer sono pieni per le pipeline che utilizzano OpenTelemetry sorgenti (otEL logs, otEL metrics e OTel trace), la pipeline inizia a restituire lo stato HTTP REQUEST_TIMEOUT con il codice di errore 408 all'endpoint di origine. Quando i buffer vengono liberati, la pipeline riprende a elaborare gli eventi.

Fonte S3

Quando i buffer sono pieni per le pipeline con una sorgente S3, le pipeline interrompono l'elaborazione delle notifiche SQS. Man mano che i buffer vengono liberati, le pipeline riprendono a elaborare le notifiche.

Se un sink è inattivo o non è in grado di inserire dati e la end-to-end conferma è abilitata per la fonte, la pipeline interrompe l'elaborazione delle notifiche SQS finché non riceve un riconoscimento riuscito da tutti i sink.

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti

Tutorial: inserisci dati in una raccolta

Creazione di pipeline