Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
Utilizzo di una pipeline di OpenSearch ingestione con Amazon S3
Con OpenSearch Ingestion, puoi utilizzare Amazon S3 come origine o come destinazione. Quando usi Amazon S3 come fonte, invii dati a una pipeline di Ingestion. OpenSearch Quando usi Amazon S3 come destinazione, scrivi dati da una pipeline di OpenSearch ingestione in uno o più bucket S3.
Amazon S3 come fonte
Esistono due modi per utilizzare Amazon S3 come fonte per elaborare i dati: con S3-SQS elaborazione e con scansioni pianificate.
Usa l' S3-SQS elaborazione quando hai bisogno di una scansione quasi in tempo reale dei file dopo che sono stati scritti su S3. Puoi configurare i bucket Amazon S3 per generare un evento ogni volta che un oggetto viene archiviato o modificato all'interno del bucket. Utilizza una scansione pianificata una tantum o ricorrente per elaborare in batch i dati in un bucket S3.
Prerequisiti
Per utilizzare Amazon S3 come fonte per una pipeline di OpenSearch Ingestion sia per una scansione pianificata che per l'elaborazione, crea innanzitutto un bucket S3. S3-SQS https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html
Nota
Se il bucket S3 utilizzato come fonte nella pipeline di OpenSearch Ingestion si trova in un altro Account AWS, devi anche abilitare le autorizzazioni di lettura tra account sul bucket. Ciò consente alla pipeline di leggere ed elaborare i dati. Per abilitare le autorizzazioni per più account, consulta la sezione Bucket owner granting cross-account bucket permissions nella Amazon S3 User Guide.
Se i tuoi bucket S3 sono in più account, usa una mappa. bucket_owners Per un esempio, vedi Cross-account S3 access
Per configurare S3-SQS l'elaborazione, è inoltre necessario eseguire i seguenti passaggi:
-
Crea una coda Amazon SQS.
-
Abilita le notifiche degli eventi sul bucket S3 con la coda SQS come destinazione.
Fase 1: Configurare il ruolo della pipeline
A differenza di altri plug-in di origine che inviano i dati a una pipeline, il plug-in di origine S3
Pertanto, affinché una pipeline possa leggere da S3, è necessario specificare un ruolo all'interno della configurazione di origine S3 della pipeline che abbia accesso sia al bucket S3 che alla coda Amazon SQS. La pipeline assumerà questo ruolo per leggere i dati dalla coda.
Nota
Il ruolo specificato nella configurazione di origine di S3 deve essere il ruolo della pipeline. Pertanto, il ruolo della pipeline deve contenere due politiche di autorizzazione separate: una per scrivere su un sink e una per estrarre dal sorgente S3. È necessario utilizzare lo stesso in tutti i componenti della pipeline. sts_role_arn
La seguente politica di esempio mostra le autorizzazioni richieste per l'utilizzo di S3 come sorgente:
È necessario allegare queste autorizzazioni al ruolo IAM specificato nell'sts_role_arnopzione all'interno della configurazione del plug-in di origine S3:
version: "2" source: s3: ... aws: ... processor: ... sink: - opensearch: ...
Passaggio 2: creare la pipeline
Dopo aver impostato le autorizzazioni, puoi configurare una pipeline di OpenSearch ingestione in base al tuo caso d'uso di Amazon S3.
S3-SQS elaborazione
Per configurare S3-SQS l'elaborazione, configura la pipeline per specificare S3 come origine e configura le notifiche Amazon SQS:
version: "2" s3-pipeline: source: s3: notification_type: "sqs" codec: newline: null sqs: queue_url: "https://sqs.us-east-1amazonaws.com/account-id/ingestion-queue" compression: "none" aws: region: "region" processor: - grok: match: message: - "%{COMMONAPACHELOG}" - date: destination: "@timestamp" from_time_received: true sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index-name" aws: region: "region"
Se osservi un basso utilizzo della CPU durante l'elaborazione di file di piccole dimensioni su Amazon S3, valuta la possibilità di aumentare il throughput modificando il valore dell'opzione. workers Per ulteriori informazioni, consulta le opzioni di configurazione del plug-in S3.
Scansione pianificata
Per impostare una scansione pianificata, configura la pipeline con una pianificazione a livello di scansione che si applica a tutti i bucket S3 o a livello di bucket. Una pianificazione a livello di bucket o una configurazione a intervalli di scansione sovrascrive sempre una configurazione a livello di scansione.
È possibile configurare le scansioni pianificate con una scansione singola, ideale per la migrazione dei dati, o una scansione ricorrente, ideale per l'elaborazione in batch.
Per configurare la pipeline per la lettura da Amazon S3, utilizza i blueprint Amazon S3 preconfigurati. Puoi modificare la scan parte della configurazione della pipeline per soddisfare le tue esigenze di pianificazione. Per ulteriori informazioni, consulta Lavorare con i progetti.
One-time scan
Una scansione pianificata viene eseguita una sola volta. Nella configurazione della pipeline, puoi utilizzare una start_time e end_time per specificare quando desideri che gli oggetti nel bucket vengano scansionati. In alternativa, puoi utilizzarlo range per specificare l'intervallo di tempo rispetto all'ora corrente in cui desideri che gli oggetti nel bucket vengano scansionati.
Ad esempio, un intervallo impostato per PT4H scansionare tutti i file creati nelle ultime quattro ore. Per configurare una scansione una tantum da eseguire una seconda volta, è necessario interrompere e riavviare la pipeline. Se non hai configurato un intervallo, devi anche aggiornare l'ora di inizio e di fine.
La configurazione seguente imposta una scansione una tantum per tutti i bucket e tutti gli oggetti in essi contenuti:
version: "2" log-pipeline: source: s3: codec: csv: compression: "none" aws: region: "region" acknowledgments: true scan: buckets: - bucket: name:my-bucketfilter: include_prefix: -Objects1/ exclude_suffix: - .jpeg - .png - bucket: name:my-bucket-2key_prefix: include: -Objects2/ exclude_suffix: - .jpeg - .png delete_s3_objects_on_read: false processor: - date: destination: "@timestamp" from_time_received: true sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index-name" aws: region: "region" dlq: s3: bucket: "dlq-bucket" region: "us-east-1"
La configurazione seguente imposta una scansione una tantum per tutti i bucket durante una finestra temporale specificata. Ciò significa che S3 elabora solo gli oggetti con tempi di creazione che rientrano in questa finestra.
scan: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z buckets: - bucket: name:my-bucket-1filter: include: -Objects1/ exclude_suffix: - .jpeg - .png - bucket: name:my-bucket-2filter: include: -Objects2/ exclude_suffix: - .jpeg - .png
La configurazione seguente imposta una scansione una tantum sia a livello di scansione che a livello di bucket. Gli orari di inizio e fine a livello di bucket sostituiscono gli orari di inizio e fine a livello di scansione.
scan: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z buckets: - bucket: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z name:my-bucket-1filter: include: -Objects1/ exclude_suffix: - .jpeg - .png - bucket: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z name:my-bucket-2filter: include: -Objects2/ exclude_suffix: - .jpeg - .png
L'arresto di una pipeline rimuove qualsiasi riferimento preesistente agli oggetti che sono stati scansionati dalla pipeline prima dell'arresto. Se una singola pipeline di scansione viene interrotta, eseguirà nuovamente la scansione di tutti gli oggetti dopo l'avvio, anche se erano già stati scansionati. Se è necessario interrompere una singola pipeline di scansione, si consiglia di modificare la finestra temporale prima di riavviare la pipeline.
Se è necessario filtrare gli oggetti in base all'ora di inizio e all'ora di fine, l'unica opzione è fermare e avviare la pipeline. Se non hai bisogno di filtrare per ora di inizio e ora di fine, puoi filtrare gli oggetti per nome. Il filtraggio per nome non richiede l'interruzione e l'avvio della pipeline. Per fare ciò, usa e. include_prefix exclude_suffix
Scansione ricorrente
Una scansione pianificata ricorrente esegue una scansione dei bucket S3 specificati a intervalli regolari e pianificati. È possibile configurare questi intervalli solo a livello di scansione perché le configurazioni a livello di bucket individuali non sono supportate.
Nella configurazione della pipeline, interval specifica la frequenza della scansione ricorrente e può essere compresa tra 30 secondi e 365 giorni. La prima di queste scansioni si verifica sempre quando si crea la pipeline. countDefinisce il numero totale di istanze di scansione.
La seguente configurazione imposta una scansione ricorrente, con un ritardo di 12 ore tra le scansioni:
scan: scheduling: interval: PT12H count: 4 buckets: - bucket: name:my-bucket-1filter: include: -Objects1/ exclude_suffix: - .jpeg - .png - bucket: name:my-bucket-2filter: include: -Objects2/ exclude_suffix: - .jpeg - .png
Amazon S3 come destinazione
Per scrivere dati da una pipeline di OpenSearch Ingestion in un bucket S3, utilizza il blueprint S3 preconfigurato per creare una pipeline con un sink S3. https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/
Quando crei il tuo sink S3, puoi specificare la tua formattazione preferita da una varietà di codec sink. https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/#codec
L'esempio seguente definisce uno schema in linea in un sink S3:
- s3: codec: parquet: schema: > { "type" : "record", "namespace" : "org.vpcFlowLog.examples", "name" : "VpcFlowLog", "fields" : [ { "name" : "version", "type" : "string"}, { "name" : "srcport", "type": "int"}, { "name" : "dstport", "type": "int"}, { "name" : "start", "type": "int"}, { "name" : "end", "type": "int"}, { "name" : "protocol", "type": "int"}, { "name" : "packets", "type": "int"}, { "name" : "bytes", "type": "int"}, { "name" : "action", "type": "string"}, { "name" : "logStatus", "type" : "string"} ] }
Quando definisci questo schema, specifica un superset di tutte le chiavi che potrebbero essere presenti nei diversi tipi di eventi che la pipeline invia a un sink.
Ad esempio, se per un evento è possibile che manchi una chiave, aggiungete quella chiave allo schema con un null valore. Le dichiarazioni di valore nullo consentono allo schema di elaborare dati non uniformi (laddove alcuni eventi hanno queste chiavi e altri no). Quando negli eventi in ingresso sono presenti queste chiavi, i relativi valori vengono scritti nei sink.
Questa definizione dello schema funge da filtro che consente solo l'invio di chiavi definite ai sink e rimuove le chiavi non definite dagli eventi in ingresso.
Puoi anche usare include_keys and exclude_keys in your sink per filtrare i dati che vengono indirizzati ad altri sink. Questi due filtri si escludono a vicenda, quindi puoi utilizzarne solo uno alla volta nello schema. Inoltre, non è possibile utilizzarli all'interno di schemi definiti dall'utente.
Per creare pipeline con tali filtri, utilizzate il modello di filtro sink preconfigurato. Per ulteriori informazioni, consulta Lavorare con i progetti.
Amazon S3 cross account come fonte
Puoi concedere l'accesso a più account con Amazon S3 in modo che le pipeline di OpenSearch Ingestion possano accedere ai bucket S3 in un altro account come fonte. Per abilitare l'accesso tra account, consulta la sezione Proprietario del bucket che concede le autorizzazioni ai bucket su più account nella Amazon S3 User Guide. Dopo aver concesso l'accesso, assicurati che il ruolo della pipeline disponga delle autorizzazioni richieste.
Quindi, puoi creare una pipeline utilizzando bucket_owners per abilitare l'accesso tra più account a un bucket Amazon S3 come fonte:
s3-pipeline: source: s3: notification_type: "sqs" codec: csv: delimiter: "," quote_character: "\"" detect_header: True sqs: queue_url: "https://sqs.ap-northeast-1.amazonaws.com/401447383613/test-s3-queue" bucket_owners: my-bucket-01: 123456789012 my-bucket-02: 999999999999 compression: "gzip"
