Uso de una canalización OpenSearch de ingestión con AWS Lambda - OpenSearch Servicio Amazon
Requisitos previos Creación de una canalizaciónAgrupación en lotesFiltrado condicional

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Uso de una canalización OpenSearch de ingestión con AWS Lambda

Utilice el AWS Lambda procesador para enriquecer los datos de cualquier fuente o destino compatible con OpenSearch Ingestion mediante un código personalizado. Con el procesador de Lambda, puede aplicar sus propias transformaciones o enriquecimientos de datos y, a continuación, devolver los eventos procesados a la canalización para su posterior procesamiento. Este procesador permite un procesamiento de datos personalizado y le brinda un control total sobre la forma en que se manipulan los datos antes de que se procesen.

nota

El límite de tamaño de carga útil para un solo evento procesado por un procesador de Lambda es de 5 MB. Además, el procesador de Lambda solo admite respuestas en formato de matriz JSON.

Requisitos previos

Antes de crear una canalización con un procesador de Lambda, cree los siguientes recursos:

  • Una AWS Lambda función que enriquece y transforma los datos de origen. Para obtener instrucciones, consulte Crear la función de Lambda.

  • Un dominio OpenSearch de servicio o una colección OpenSearch sin servidor que será el colector de la canalización. Para obtener más información, consulte Creación de dominios OpenSearch de servicio y Creación de colecciones.

  • Un rol de canalización que incluye permisos para escribir en el dominio o receptor de la colección. Para obtener más información, consulte Rol de canalización.

    El rol de canalización también necesita una política de permisos adjunta que le permita invocar la función de Lambda especificada en la configuración de canalización. Por ejemplo:

    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"
            
        }
    ]
}

Creación de una canalización

Para usarlo AWS Lambda como procesador, configure una canalización de OpenSearch ingestión y especifíquela aws_lambda como procesador. También puede usar el esquema de enriquecimiento de AWS Lambda personalizado para crear la canalización. Para obtener más información, consulte Uso de esquemas.

El siguiente ejemplo de canalización recibe datos de una fuente HTTP, los enriquece con un procesador de datos y el AWS Lambda procesador e ingiere los datos procesados en un dominio. OpenSearch 

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

La siguiente AWS Lambda función de ejemplo transforma los datos entrantes añadiendo un nuevo par clave-valor ("transformed": "true") a cada elemento de la matriz de eventos proporcionada y, a continuación, devuelve la versión modificada.

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

Agrupación en lotes

Las canalizaciones envían eventos por lotes al procesador de Lambda y ajustan dinámicamente el tamaño del lote para garantizar que se mantenga por debajo del límite de 5 MB.

El siguiente es un ejemplo de lote de canalización:

batch:
    key_name: "events"
    
input_arrary = event.get('events', [])
nota

Al crear una canalización, asegúrese de que la opción key_name de la configuración del procesador de Lambda coincida con la clave de evento del controlador de Lambda.

Filtrado condicional

El filtrado condicional le permite controlar cuándo el AWS Lambda procesador invoca la función Lambda en función de condiciones específicas de los datos de eventos. Esto resulta especialmente útil cuando desea procesar de forma selectiva ciertos tipos de eventos e ignorar otros.

El siguiente ejemplo de configuración utiliza el filtrado condicional:

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