Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Utilisation d'un pipeline OpenSearch d'ingestion avec Amazon S3

Avec OpenSearch Ingestion, vous pouvez utiliser Amazon S3 comme source ou comme destination. Lorsque vous utilisez Amazon S3 comme source, vous envoyez des données à un pipeline d' OpenSearch ingestion. Lorsque vous utilisez Amazon S3 comme destination, vous écrivez les données d'un pipeline d' OpenSearch ingestion vers un ou plusieurs compartiments S3.

Amazon S3 en tant que source

Vous pouvez utiliser Amazon S3 comme source pour traiter les données de deux manières : avec S3-SQS traitement et avec des scans planifiés.

Utilisez S3-SQS le traitement lorsque vous avez besoin d'une analyse en temps quasi réel des fichiers une fois qu'ils ont été écrits sur S3. Vous pouvez configurer des compartiments Amazon S3 pour déclencher un événement chaque fois qu'un objet est stocké ou modifié dans le compartiment. Utilisez une analyse planifiée ponctuelle ou récurrente pour traiter par lots les données d'un compartiment S3.

Conditions préalables

Pour utiliser Amazon S3 comme source d'un pipeline d' OpenSearch ingestion à la fois pour un scan ou un S3-SQS traitement planifiés, créez d'abord un compartiment S3.

Pour configurer S3-SQS le traitement, vous devez également suivre les étapes suivantes :

Étape 1 : configurer le rôle du pipeline

Contrairement aux autres plug-ins sources qui envoient des données vers un pipeline, le plug-in source S3 possède une architecture basée sur la lecture dans laquelle le pipeline extrait les données de la source.

Par conséquent, pour qu'un pipeline puisse lire depuis S3, vous devez spécifier un rôle dans la configuration source S3 du pipeline qui a accès à la fois au compartiment S3 et à la file d'attente Amazon SQS. Le pipeline assumera ce rôle afin de lire les données de la file d'attente.

L'exemple de politique suivant indique les autorisations requises pour utiliser S3 en tant que source :

JSON

{
  "Version":"2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action":[
          "s3:ListBucket",
          "s3:GetBucketLocation",
          "s3:GetObject"
       ],
      "Resource": "arn:aws:s3:::amzn-s3-demo-bucket/*"
    },
    {
       "Effect":"Allow",
       "Action":"s3:ListAllMyBuckets",
       "Resource":"arn:aws:s3:::*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:DeleteMessage",
        "sqs:ReceiveMessage",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": "arn:aws:sqs:us-east-1:111122223333:MyS3EventSqsQueue"
    }
  ]
}

Vous devez associer ces autorisations au rôle IAM que vous spécifiez dans l'sts_role_arnoption de la configuration du plug-in source S3 :

version: "2"
source:
  s3:
    ...
    aws:
      ...
processor:
  ...
sink:
  - opensearch:
      ...

Étape 2 : Création du pipeline

Après avoir configuré vos autorisations, vous pouvez configurer un pipeline d' OpenSearch ingestion en fonction de votre cas d'utilisation d'Amazon S3.

S3-SQS traitement

Pour configurer S3-SQS le traitement, configurez votre pipeline pour spécifier S3 comme source et configurez les notifications 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"

Si vous observez une faible utilisation du processeur lors du traitement de petits fichiers sur Amazon S3, envisagez d'augmenter le débit en modifiant la valeur de l'workersoption. Pour plus d'informations, consultez les options de configuration du plug-in S3.

Scan programmé

Pour configurer un scan programmé, configurez votre pipeline avec un calendrier au niveau du scan qui s'applique à tous vos compartiments S3, ou au niveau du bucket. Une planification au niveau du compartiment ou une configuration à intervalles de numérisation remplacent toujours une configuration au niveau du scan.

Vous pouvez configurer des analyses planifiées soit avec une analyse unique, idéale pour la migration des données, soit avec une analyse récurrente, idéale pour le traitement par lots.

Pour configurer votre pipeline afin qu'il puisse lire depuis Amazon S3, utilisez les plans Amazon S3 préconfigurés. Vous pouvez modifier la scan partie de la configuration de votre pipeline pour répondre à vos besoins de planification. Pour de plus amples informations, veuillez consulter Travailler avec des plans.

One-time scan

Une analyse planifiée unique est exécutée une seule fois. Dans la configuration de votre pipeline, vous pouvez utiliser un start_time et end_time pour spécifier à quel moment vous souhaitez que les objets du compartiment soient scannés. Vous pouvez également spécifier l'rangeintervalle de temps par rapport à l'heure actuelle pendant lequel vous souhaitez que les objets du compartiment soient scannés.

Par exemple, une plage définie pour analyser PT4H tous les fichiers créés au cours des quatre dernières heures. Pour configurer un scan unique afin qu'il soit exécuté une seconde fois, vous devez arrêter et redémarrer le pipeline. Si aucune plage n'est configurée, vous devez également mettre à jour les heures de début et de fin.

La configuration suivante met en place un scan unique pour tous les compartiments et tous les objets qu'ils contiennent :

version: "2"
log-pipeline:
  source:
    s3:
      codec:
        csv:
      compression: "none"
      aws:
        region: "region"
      acknowledgments: true
      scan:
        buckets:
          - bucket:
              name: my-bucket
              filter:
                include_prefix:
                  - Objects1/
                exclude_suffix:
                  - .jpeg
                  - .png
          - bucket:
              name: my-bucket-2
              key_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 configuration suivante met en place une analyse unique pour tous les compartiments pendant une fenêtre de temps spécifiée. Cela signifie que S3 traite uniquement les objets dont les heures de création se situent dans cette fenêtre.

scan:
  start_time: 2023-01-21T18:00:00.000Z
  end_time: 2023-04-21T18:00:00.000Z
  buckets:
    - bucket:
        name: my-bucket-1
        filter:
          include:
            - Objects1/
          exclude_suffix:
            - .jpeg
            - .png
    - bucket:
        name: my-bucket-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

La configuration suivante permet de configurer un scan unique à la fois au niveau du scan et au niveau du bucket. Les heures de début et de fin au niveau du compartiment remplacent les heures de début et de fin au niveau du scan.

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-1
        filter:
          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-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

L'arrêt d'un pipeline supprime toute référence préexistante aux objets qui ont été scannés par le pipeline avant l'arrêt. Si un seul pipeline de numérisation est arrêté, il réanalysera tous les objets après son démarrage, même s'ils ont déjà été scannés. Si vous devez arrêter un seul pipeline de scan, il est recommandé de modifier votre fenêtre de temps avant de redémarrer le pipeline.

Si vous devez filtrer les objets par heure de début et de fin, arrêter et démarrer votre pipeline est la seule option. Si vous n'avez pas besoin de filtrer par heure de début et heure de fin, vous pouvez filtrer les objets par nom. Le fait de filtrer par nom ne vous oblige pas à arrêter et à démarrer votre pipeline. Pour ce faire, utilisez include_prefix etexclude_suffix.

Scan récurrent

Une analyse planifiée récurrente analyse les compartiments S3 que vous avez spécifiés à des intervalles réguliers et planifiés. Vous ne pouvez configurer ces intervalles qu'au niveau du scan, car les configurations individuelles au niveau du bucket ne sont pas prises en charge.

Dans la configuration de votre pipeline, interval cela spécifie la fréquence du scan récurrent, qui peut être comprise entre 30 secondes et 365 jours. La première de ces analyses a toujours lieu lorsque vous créez le pipeline. countDéfinit le nombre total d'instances de scan.

La configuration suivante met en place un scan récurrent, avec un délai de 12 heures entre les scans :

scan:
  scheduling:
    interval: PT12H
    count: 4
  buckets:
    - bucket:
        name: my-bucket-1
        filter:
          include:
            - Objects1/
          exclude_suffix:
            - .jpeg
            - .png
    - bucket:
        name: my-bucket-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

Amazon S3 comme destination

Pour écrire des données depuis un pipeline d' OpenSearch ingestion vers un compartiment S3, utilisez le plan S3 préconfiguré pour créer un pipeline avec un récepteur S3. Ce pipeline achemine les données sélectives vers un OpenSearch récepteur et envoie simultanément toutes les données pour archivage dans S3. Pour de plus amples informations, veuillez consulter Travailler avec des plans.

Lorsque vous créez votre récepteur S3, vous pouvez spécifier votre format préféré à partir de divers codecs de récepteur. Par exemple, si vous souhaitez écrire des données au format colonnaire, choisissez le codec Parquet ou Avro. Si vous préférez un format basé sur les lignes, choisissez JSON ou NDJSON. Pour écrire des données dans S3 dans un schéma spécifié, vous pouvez également définir un schéma en ligne dans des codecs récepteurs à l'aide du format Avro.

L'exemple suivant définit un schéma en ligne dans un récepteur 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"}
           ]
         }

Lorsque vous définissez ce schéma, spécifiez un sur-ensemble de toutes les clés susceptibles d'être présentes dans les différents types d'événements que votre pipeline transmet à un récepteur.

Par exemple, s'il est possible qu'une clé soit manquante pour un événement, ajoutez cette clé dans votre schéma avec une null valeur. Les déclarations de valeurs nulles permettent au schéma de traiter des données non uniformes (où certains événements possèdent ces clés et d'autres non). Lorsque ces clés sont présentes pour les événements entrants, leurs valeurs sont écrites dans des cuves.

Cette définition de schéma agit comme un filtre qui permet uniquement d'envoyer des clés définies aux récepteurs et supprime les clés non définies des événements entrants.

Vous pouvez également utiliser include_keys et exclude_keys dans votre évier pour filtrer les données qui sont acheminées vers d'autres cuves. Ces deux filtres s'excluent mutuellement, vous ne pouvez donc en utiliser qu'un à la fois dans votre schéma. De plus, vous ne pouvez pas les utiliser dans des schémas définis par l'utilisateur.

Pour créer des canalisations avec de tels filtres, utilisez le schéma de filtre d'évier préconfiguré. Pour de plus amples informations, veuillez consulter Travailler avec des plans.

Compte croisé Amazon S3 en tant que source

Vous pouvez accorder l'accès à tous les comptes avec Amazon S3 afin que les pipelines OpenSearch d'ingestion puissent accéder aux compartiments S3 d'un autre compte en tant que source. Pour activer l'accès entre comptes, consultez la section Le propriétaire du bucket accorde des autorisations de bucket à plusieurs comptes dans le guide de l'utilisateur Amazon S3. Après avoir accordé l'accès, assurez-vous que votre rôle de pipeline dispose des autorisations requises.

Vous pouvez ensuite créer un pipeline en utilisant bucket_owners pour activer l'accès entre comptes à un compartiment Amazon S3 en tant que source :

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"