Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
Verwenden einer OpenSearch Ingestion-Pipeline mit Amazon S3
Mit OpenSearch Ingestion können Sie Amazon S3 als Quelle oder als Ziel verwenden. Wenn Sie Amazon S3 als Quelle verwenden, senden Sie Daten an eine OpenSearch Ingestion-Pipeline. Wenn Sie Amazon S3 als Ziel verwenden, schreiben Sie Daten aus einer OpenSearch Ingestion-Pipeline in einen oder mehrere S3-Buckets.
Amazon S3 als Quelle
Es gibt zwei Möglichkeiten, Amazon S3 als Quelle für die Verarbeitung von Daten zu verwenden — mit der S3-SQS Verarbeitung und mit geplanten Scans.
Verwenden Sie S3-SQS Processing, wenn Sie Dateien nahezu in Echtzeit scannen müssen, nachdem sie in S3 geschrieben wurden. Sie können Amazon S3-Buckets so konfigurieren, dass jedes Mal, wenn ein Objekt im Bucket gespeichert oder geändert wird, ein Ereignis ausgelöst wird. Verwenden Sie einen einmaligen oder wiederkehrenden geplanten Scan, um Daten in einem S3-Bucket stapelweise zu verarbeiten.
Themen
Voraussetzungen
Um Amazon S3 als Quelle für eine OpenSearch Ingestion-Pipeline sowohl für einen geplanten Scan als auch für eine S3-SQS Verarbeitung zu verwenden, erstellen Sie zunächst einen S3-Bucket.
Anmerkung
Wenn sich der als Quelle in der OpenSearch Ingestion-Pipeline verwendete S3-Bucket in einem anderen befindet AWS-Konto, müssen Sie auch kontoübergreifende Leseberechtigungen für den Bucket aktivieren. Dadurch kann die Pipeline die Daten lesen und verarbeiten. Informationen zur Aktivierung kontoübergreifender Berechtigungen finden Sie im Amazon S3-Benutzerhandbuch unter Bucket-Besitzer, der kontoübergreifende Bucket-Berechtigungen gewährt.
Wenn sich Ihre S3-Buckets in mehreren Konten befinden, verwenden Sie eine Map. bucket_owners Ein Beispiel finden Sie in der OpenSearch Dokumentation unter
Um die S3-SQS Verarbeitung einzurichten, müssen Sie außerdem die folgenden Schritte ausführen:
-
Aktivieren Sie Ereignisbenachrichtigungen im S3-Bucket mit der SQS-Warteschlange als Ziel.
Schritt 1: Konfigurieren Sie die Pipeline-Rolle
Im Gegensatz zu anderen Quell-Plugins, die Daten an eine Pipeline übertragen, verfügt das
Damit eine Pipeline aus S3 lesen kann, müssen Sie daher eine Rolle in der S3-Quellkonfiguration der Pipeline angeben, die Zugriff sowohl auf den S3-Bucket als auch auf die Amazon SQS-Warteschlange hat. Die Pipeline übernimmt diese Rolle, um Daten aus der Warteschlange zu lesen.
Anmerkung
Die Rolle, die Sie in der S3-Quellkonfiguration angeben, muss die Pipeline-Rolle sein. Daher muss Ihre Pipeline-Rolle zwei separate Berechtigungsrichtlinien enthalten — eine zum Schreiben in eine Senke und eine zum Abrufen aus der S3-Quelle. Sie müssen dasselbe sts_role_arn in allen Pipeline-Komponenten verwenden.
Die folgende Beispielrichtlinie zeigt die erforderlichen Berechtigungen für die Verwendung von S3 als Quelle:
Sie müssen diese Berechtigungen an die IAM-Rolle anhängen, die Sie in der sts_role_arn Option in der Konfiguration des S3-Quell-Plug-ins angeben:
version: "2" source: s3: ... aws: ... processor: ... sink: - opensearch: ...
Schritt 2: Erstellen Sie die Pipeline
Nachdem Sie Ihre Berechtigungen eingerichtet haben, können Sie je nach Ihrem Amazon OpenSearch S3-Anwendungsfall eine Ingestion-Pipeline konfigurieren.
S3-SQS Verarbeitung
Um die S3-SQS Verarbeitung einzurichten, konfigurieren Sie Ihre Pipeline so, dass S3 als Quelle angegeben wird, und richten Sie Amazon SQS-Benachrichtigungen ein:
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"
Wenn Sie bei der Verarbeitung kleiner Dateien auf Amazon S3 eine geringe CPU-Auslastung feststellen, sollten Sie erwägen, den Durchsatz zu erhöhen, indem Sie den Wert der workers Option ändern. Weitere Informationen finden Sie in den Konfigurationsoptionen des S3-Plugins
Geplanter Scan
Um einen geplanten Scan einzurichten, konfigurieren Sie Ihre Pipeline mit einem Zeitplan auf Scan-Ebene, der für alle Ihre S3-Buckets gilt, oder auf Bucket-Ebene. Ein Zeitplan auf Bucket-Ebene oder eine Scan-Intervallkonfiguration überschreibt immer eine Konfiguration auf Scan-Ebene.
Sie können geplante Scans entweder mit einem einmaligen Scan konfigurieren, der sich ideal für die Datenmigration eignet, oder mit einem wiederkehrenden Scan, der sich ideal für die Stapelverarbeitung eignet.
Verwenden Sie die vorkonfigurierten Amazon S3-Blueprints, um Ihre Pipeline so zu konfigurieren, dass sie aus Amazon S3 liest. Sie können den scan Teil Ihrer Pipeline-Konfiguration bearbeiten, um Ihren Planungsanforderungen zu entsprechen. Weitere Informationen finden Sie unter Mit Blueprints arbeiten.
One-time scan
Ein einmaliger geplanter Scan wird einmal ausgeführt. In Ihrer Pipeline-Konfiguration können Sie ein start_time und verwenden, end_time um anzugeben, wann die Objekte im Bucket gescannt werden sollen. Alternativ können range Sie das Zeitintervall im Verhältnis zur aktuellen Uhrzeit angeben, in dem die Objekte im Bucket gescannt werden sollen.
Beispiel: Ein Bereich, der so eingestellt ist, dass alle Dateien PT4H gescannt werden, die in den letzten vier Stunden erstellt wurden. Um einen einmaligen Scan so zu konfigurieren, dass er ein zweites Mal ausgeführt wird, müssen Sie die Pipeline beenden und neu starten. Wenn Sie keinen Bereich konfiguriert haben, müssen Sie auch die Start- und Endzeiten aktualisieren.
Die folgende Konfiguration richtet einen einmaligen Scan für alle Buckets und alle Objekte in diesen Buckets ein:
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"
Die folgende Konfiguration richtet einen einmaligen Scan für alle Buckets während eines bestimmten Zeitfensters ein. Das bedeutet, dass S3 nur die Objekte verarbeitet, deren Erstellungszeiten in dieses Fenster fallen.
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
Die folgende Konfiguration richtet einen einmaligen Scan sowohl auf Scanebene als auch auf Bucket-Ebene ein. Start- und Endzeiten auf Bucket-Ebene überschreiben Start- und Endzeiten auf Scan-Ebene.
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
Beim Stoppen einer Pipeline werden alle bereits vorhandenen Hinweise darauf entfernt, welche Objekte vor dem Stopp von der Pipeline gescannt wurden. Wenn eine einzelne Scan-Pipeline gestoppt wird, scannt sie alle Objekte nach dem Start erneut, auch wenn sie bereits gescannt wurden. Wenn Sie eine einzelne Scan-Pipeline stoppen müssen, wird empfohlen, Ihr Zeitfenster zu ändern, bevor Sie die Pipeline erneut starten.
Wenn Sie Objekte nach Start- und Endzeit filtern müssen, ist das Stoppen und Starten der Pipeline die einzige Option. Wenn Sie nicht nach Start- und Endzeit filtern müssen, können Sie Objekte nach Namen filtern. Für das Filtern nach Namen müssen Sie Ihre Pipeline nicht beenden und starten. Verwenden include_prefix Sie dazu und. exclude_suffix
Wiederkehrender Scan
Bei einem wiederkehrenden geplanten Scan werden Ihre angegebenen S3-Buckets in regelmäßigen, geplanten Intervallen gescannt. Sie können diese Intervalle nur auf Scan-Ebene konfigurieren, da individuelle Konfigurationen auf Bucket-Ebene nicht unterstützt werden.
In Ihrer Pipeline-Konfiguration interval gibt der die Häufigkeit des wiederkehrenden Scans an und kann zwischen 30 Sekunden und 365 Tagen liegen. Der erste dieser Scans findet immer statt, wenn Sie die Pipeline erstellen. Das count definiert die Gesamtzahl der Scan-Instanzen.
Mit der folgenden Konfiguration wird ein wiederkehrender Scan mit einer Verzögerung von 12 Stunden zwischen den Scans eingerichtet:
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 als Ziel
Um Daten aus einer OpenSearch Ingestion-Pipeline in einen S3-Bucket zu schreiben, verwenden Sie den vorkonfigurierten S3-Blueprint, um eine Pipeline mit einer S3-Senke zu erstellen. https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/
Wenn Sie Ihre S3-Senke erstellen, können Sie Ihre bevorzugte Formatierung anhand einer Vielzahl von https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/#codec
Das folgende Beispiel definiert ein Inline-Schema in einer S3-Senke:
- 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"} ] }
Wenn Sie dieses Schema definieren, geben Sie eine Obermenge aller Schlüssel an, die in den verschiedenen Ereignistypen, die Ihre Pipeline an eine Senke sendet, vorkommen können.
Wenn bei einem Ereignis beispielsweise die Möglichkeit besteht, dass ein Schlüssel fehlt, fügen Sie diesen Schlüssel Ihrem Schema mit einem null Wert hinzu. Nullwertdeklarationen ermöglichen es dem Schema, uneinheitliche Daten zu verarbeiten (wobei einige Ereignisse diese Schlüssel haben und andere nicht). Wenn bei eingehenden Ereignissen diese Schlüssel vorhanden sind, werden ihre Werte in Senken geschrieben.
Diese Schemadefinition fungiert als Filter, der nur das Senden definierter Schlüssel an Senken zulässt und undefinierte Schlüssel aus eingehenden Ereignissen entfernt.
Sie können auch include_keys und exclude_keys in Ihrer Senke verwenden, um Daten zu filtern, die an andere Senken weitergeleitet werden. Diese beiden Filter schließen sich gegenseitig aus, sodass Sie in Ihrem Schema jeweils nur einen Filter verwenden können. Außerdem können Sie sie nicht in benutzerdefinierten Schemas verwenden.
Verwenden Sie den vorkonfigurierten Senkenfilter-Blueprint, um Pipelines mit solchen Filtern zu erstellen. Weitere Informationen finden Sie unter Mit Blueprints arbeiten.
Amazon S3 Cross Account als Quelle
Sie können kontoübergreifenden Zugriff mit Amazon S3 gewähren, sodass OpenSearch Ingestion-Pipelines auf S3-Buckets in einem anderen Konto als Quelle zugreifen können. Informationen zum Ermöglichen des kontoübergreifenden Zugriffs finden Sie im Amazon S3-Benutzerhandbuch unter Bucket-Besitzer, der kontoübergreifende Bucket-Berechtigungen gewährt. Nachdem Sie den Zugriff gewährt haben, stellen Sie sicher, dass Ihre Pipeline-Rolle über die erforderlichen Berechtigungen verfügt.
Anschließend können Sie eine Pipeline erstellen, bucket_owners um den kontoübergreifenden Zugriff auf einen Amazon S3-Bucket als Quelle zu ermöglichen:
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"
