amazon web services - que - ¿Cómo procesar la cola SQS con la función lambda(no a través de eventos programados)?

sqs que es

Question

Aquí está el esquema simplificado que estoy tratando de hacer funcionar:

solicitudes http -> (Gateway API + lambda A) -> SQS -> (lambda B ?????) -> DynamoDB

Por lo tanto, debería funcionar como se muestra: los datos provenientes de muchas solicitudes http (hasta 500 por segundo, por ejemplo) se colocan en la cola SQS por mi función lambda A. Luego, la otra función, B, procesa la cola: lee hasta 10 elementos (de forma periódica) y los escribe en DynamoDB con BatchWriteItem.

El problema es que no puedo entender cómo activar la segunda función lambda. Debe llamarse con frecuencia, varias veces por segundo (o al menos una vez por segundo), porque necesito todos los datos de la cola para ingresar a DynamoDB lo antes posible (es por eso que llamar a la función lambda B a través de eventos programados como se describe here no es una opción )

¿Por qué no quiero escribir directamente en DynamoDB, sin SQS?

Sería genial para mí evitar el uso de SQS. El problema que estoy tratando de resolver con SQS es la limitación de DynamoDB. Ni siquiera se estrangula sino la forma en que se maneja al escribir datos en DynamoDB con AWS SDK: al escribir registros uno por uno y estrangularlos, AWS SDK vuelve a intentar escribir en silencio, lo que resulta en un aumento del tiempo de procesamiento de solicitudes desde el punto de ver.

Por lo tanto, me gustaría almacenar temporalmente los datos en la cola, enviar la respuesta "200 OK" de regreso al cliente y luego hacer que la cola sea procesada por una función separada, escribiendo múltiples registros con una llamada BatchWriteItem de DynamoDB (que devuelve elementos no procesados ​​en lugar de reintento automático en caso de que de estrangulamiento). Incluso preferiría perder algunos registros en lugar de aumentar el retraso entre la recepción y el almacenamiento de un registro en DynamoDB

UPD: Si alguien está interesado, he encontrado cómo hacer que aws-sdk omita los reintentos automáticos en caso de limitación: hay un parámetro especial maxRetries . De todos modos, voy a usar Kinesis como se sugiere a continuación

Answer 1

¿Quizás una solución más rentable sería mantener todo en el SQS (tal como está) y luego ejecutar un evento programado que invoca una función Lambda de subprocesos múltiples que procesa los elementos de la cola?

De esta manera, su trabajador de cola puede coincidir exactamente con sus límites. Si la cola está vacía, la función puede finalizar prematuramente o comenzar a sondear en un solo hilo.

Kinesis suena como una matanza excesiva para este caso: no necesita el pedido original, por ejemplo. Además, ejecutar múltiples Lambdas simultáneamente es seguramente más costoso que ejecutar solo un Lambda multiproceso.

Su Lambda tendrá que ver con E / S, haciendo llamadas externas a los servicios de AWS, por lo que una función puede encajar muy bien.

Answer 2

A partir del 28 de junio de 2018, ahora puede usar SQS para activar las funciones de AWS Lambda de forma nativa. ¡Ya no se necesitan soluciones alternativas!

https://aws.amazon.com/blogs/aws/aws-lambda-adds-amazon-simple-queue-service-to-supported-event-sources/

Answer 3

Aquí está mi solución a este problema:

HTTP request --> DynamoDb --> Stream --> Lambda Function

En esta solución, debe configurar una secuencia para la tabla. La transmisión se maneja con una función Lambda que escribirás y eso es todo. No es necesario usar SQS ni nada más.

Por supuesto, este es un diseño simplificado y funciona solo para problemas simples. Para escenarios más complicados, use Kinesis (como se menciona en las otras respuestas).

Aquí hay un enlace a la documentación de AWS sobre el tema .

Answer 4

Así es como recopilo mensajes de una cola SQS:

package au.com.redbarn.aws.lambda2lambda_via_sqs;

import java.util.List;

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;

import lombok.extern.log4j.Log4j2;

@Log4j2
public class SQSConsumerLambda implements RequestHandler<SQSEvent, String> {

    @Override
    public String handleRequest(SQSEvent input, Context context) {

        log.info("message received");

        List<SQSMessage> records = input.getRecords();

        for (SQSMessage record : records) {
            log.info(record.getBody());
        }

        return "Ok";
    }
}

Agregue su código DynamoDB para handleRequest() y Lambda B estará listo.

Answer 5

Creo que AWS ahora ha encontrado una manera en la que SQS puede desencadenar una función lambda. Así que supongo que podemos usar SQS para suavizar grandes cantidades de datos a dinamo en caso de que no te importe el orden de los mensajes. Consulte su blog en esta nueva actualización: https://aws.amazon.com/blogs/aws/aws-lambda-adds-amazon-simple-queue-service-to-supported-event-sources/

Answer 6

Desafortunadamente, no puede hacer esto integrando directamente SQS y Lambda. Pero no te preocupes demasiado todavía. ¡Hay una solucion! Debe agregar otro servicio de Amazon a la mezcla y todos sus problemas se resolverán.

http requests --> (Gateway API + lambda A) --> SQS + SNS --> lambda B --> DynamoDB

Puede activar una notificación SNS al segundo servicio lambda para iniciarla. Una vez que se inicia, puede drenar la cola y escribir todos los resultados en DynamoDB. Para comprender mejor las posibles fuentes de eventos para Lambda, consulte estos documentos .

Answer 7

Otra solución sería simplemente agregar el elemento a SQS, llamar a la función Lambda objetivo con Event para que sea asíncrono.

El Lambda asíncrono puede obtener de SQS tantos elementos como desee y procesarlos.

También agregaría una llamada programada a la Lambda asíncrona para manejar cualquier elemento en la cola que haya tenido un error.

[ACTUALIZACIÓN] Ahora puede configurar el activador Lambda en un nuevo mensaje en la cola

Answer 8

[Esto no responde directamente a su pregunta explícita, por lo que en mi experiencia será rechazada :) Sin embargo, responderé el problema fundamental que está tratando de resolver.]

La forma en que tomamos una avalancha de solicitudes entrantes y las enviamos a las funciones de AWS Lambda para escribir de forma dinámica en DynamoDB es reemplazar SQS en la arquitectura propuesta con transmisiones de Amazon Kinesis.

Las transmisiones de Kinesis pueden controlar las funciones de AWS Lambda.

Las transmisiones de Kinesis garantizan el orden de los mensajes entregados para cualquier clave dada (ideal para operaciones de base de datos ordenadas).

Las transmisiones de Kinesis le permiten especificar cuántas funciones de AWS Lambda se pueden ejecutar en paralelo (una por partición), lo que se puede coordinar con su capacidad de escritura DynamoDB.

Las transmisiones de Kinesis pueden pasar varios mensajes disponibles en una invocación de la función AWS Lambda, lo que permite una mayor optimización.

Nota: Es realmente el servicio AWS Lambda el que lee de las transmisiones de Amazon Kinesis y luego invoca la función, y no las transmisiones de Kinesis que invocan directamente a AWS Lambda; pero a veces es más fácil visualizarlo mientras Kinesis lo conduce. El resultado para el usuario es casi el mismo.