View a markdown version of this page

Utilisation d'un pipeline OpenSearch d'ingestion avec AWS Lambda - OpenSearch Service Amazon
Conditions préalablesCréation d’un pipelineTraitement par lotFiltrage conditionnel

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 AWS Lambda

Utilisez le AWS Lambda processeur pour enrichir les données provenant de n'importe quelle source ou destination prise en charge par OpenSearch Ingestion à l'aide d'un code personnalisé. Avec le processeur Lambda, vous pouvez appliquer vos propres transformations ou enrichissements de données, puis renvoyer les événements traités à votre pipeline pour un traitement ultérieur. Ce processeur permet un traitement personnalisé des données et vous donne un contrôle total sur la façon dont les données sont manipulées avant leur passage dans le pipeline.

Note

La limite de taille de charge utile pour un seul événement traité par un processeur Lambda est de 5 Mo. En outre, le processeur Lambda ne prend en charge que les réponses au format tableau JSON.

Conditions préalables

Avant de créer un pipeline avec un processeur Lambda, créez les ressources suivantes :

  • Une AWS Lambda fonction qui enrichit et transforme vos données sources. Pour obtenir des instructions, voir Création de votre première fonction Lambda.

  • Un domaine OpenSearch de service ou une collection OpenSearch Serverless qui servira de récepteur de pipeline. Pour plus d’informations, consultez Création de domaines OpenSearch de service et Créer des collections.

  • Un rôle de pipeline qui inclut les autorisations d'écriture dans le domaine ou le collecteur de collecte. Pour de plus amples informations, veuillez consulter Rôle du pipeline.

    Le rôle de pipeline nécessite également une politique d'autorisations associée qui lui permet d'invoquer la fonction Lambda spécifiée dans la configuration du pipeline. Par exemple :

    JSON
    {
    "Version":"2012-10-17",
    "Statement": [
        {
            "Sid": "allowinvokeFunction",
            "Effect": "Allow",
            "Action": [
                "lambda:invokeFunction",
                "lambda:InvokeAsync",
                "lambda:ListFunctions"
            ],
            "Resource": "arn:aws:lambda:us-east-1:111122223333:function:function-name"
            
        }
    ]
}

Création d’un pipeline

Pour l'utiliser AWS Lambda en tant que processeur, configurez un pipeline d' OpenSearch ingestion et aws_lambda spécifiez-le en tant que processeur. Vous pouvez également utiliser le plan d'enrichissement AWS Lambda personnalisé pour créer le pipeline. Pour de plus amples informations, veuillez consulter Travailler avec des plans.

L'exemple de pipeline suivant reçoit des données d'une source HTTP, les enrichit à l'aide d'un processeur de date et du AWS Lambda processeur, et ingère les données traitées vers un OpenSearch domaine.

version: "2"
lambda-processor-pipeline:
  source:
    http:
      path: "/${pipelineName}/logs"
  processor:
      - date:
        destination: "@timestamp"
        from_time_received: true
    - aws_lambda:
        function_name: "my-lambda-function"

        tags_on_failure: ["lambda_failure"]
        batch:
            key_name: "events"
        aws:
          region: region
  sink:
    - opensearch:
        hosts: [ "https://search-mydomain.us-east-1es.amazonaws.com" ]
        index: "table-index"
        aws:
          region: "region"
          serverless: false

L'exemple de AWS Lambda fonction suivant transforme les données entrantes en ajoutant une nouvelle paire clé-valeur ("transformed": "true") à chaque élément du tableau d'événements fourni, puis renvoie la version modifiée.

import json

def lambda_handler(event, context):
    input_array = event.get('events', [])
    output = []
    for input in input_array:
        input["transformed"] = "true";
        output.append(input)

    return output

Traitement par lot

Les pipelines envoient des événements par lots au processeur Lambda et ajustent dynamiquement la taille du lot pour s'assurer qu'elle reste inférieure à la limite de 5 Mo.

Voici un exemple de lot de pipeline :

batch:
    key_name: "events"

input_arrary = event.get('events', [])
Note

Lorsque vous créez un pipeline, assurez-vous que l'key_nameoption de la configuration du processeur Lambda correspond à la clé d'événement du gestionnaire Lambda.

Filtrage conditionnel

Le filtrage conditionnel vous permet de contrôler le moment où votre AWS Lambda processeur invoque la fonction Lambda en fonction de conditions spécifiques dans les données d'événements. Cela est particulièrement utile lorsque vous souhaitez traiter certains types d'événements de manière sélective tout en ignorant les autres.

L'exemple de configuration suivant utilise le filtrage conditionnel :

processors:
  - aws_lambda:
      function_name: "my-lambda-function"
      aws:
        region: "region"
      lambda_when: "/sourceIp == 10.10.10.10"