Para agentes de IA: hay un Ć­ndice de documentaciĆ³n disponible en https://www.mongodb.com/es/docs/llms.txt ā€” versiones en markdown de todas las pĆ”ginas estĆ”n disponibles agregando .md a cualquier ruta URL.
Docs Menu
Docs Home
/ /
Etapas de agregaciĆ³n
Docs Home
/
Desarrollo
/
Atlas Stream Processing
/
Etapas de agregaciĆ³n
Docs Home
/
Desarrollo
/
Atlas Stream Processing
/
Etapas de agregaciĆ³n

$externalFunction

DefiniciĆ³n

$externalFunction

La etapa $externalFunction activa procesos en un recurso especĆ­fico de AWS Lambda. Su solicitud al proceso AWS Lambda puede ser sincrĆ³nica o asincrĆ³nica.

Crea una funciĆ³n AWS Lambda y autentĆ­cate con Unified AWS Access

Para llamar a un recurso de AWS Lambda desde dentro de tu pipeline de Atlas Stream Processing, tu AWS Lambda debe estar implementado en la misma regiĆ³n de AWS en la que tu Atlas Stream Processing estĆ© implementado. Para obtener mĆ”s informaciĆ³n sobre cĆ³mo implementar un recurso de AWS Lambda, consulta la documentaciĆ³n de AWS.

1

Crea una funciĆ³n de AWS Lambda.

Con la CLI de AWS o a travĆ©s de la Interfaz de Usuario de AWS, crea una funciĆ³n lambda.

Nota

  • Los recursos de AWS Lambda deben crearse con un tipo de autenticaciĆ³n AWS_IAM, que requiere que crees un rol y una polĆ­ticade AWS IAM.

  • Solo se admite el tipo de respuesta en bĆŗfer, no de transmisiĆ³n.

2

Configurar acceso unificado a AWS.

Nota

El procedimiento que se describe aquĆ­ solo cubre el flujo bĆ”sico de configuraciĆ³n en la interfaz de usuario de Atlas. Para aprender mĆ”s, consultar la documentaciĆ³n Configurar el acceso unificado a AWS.

Acceso requerido

Para configurar el acceso unificado a AWS, debes tener acceso a Organization Owner o a Project Owner en el proyecto.

Requisitos previos

Nota

  • Su polĆ­tica de AWS IAM debe incluir la acciĆ³n lambda:InvokeFunction.

  • Debe remplazar los valores de marcador de posiciĆ³n ExternalId y Resource con los suyos propios, que estĆ”n disponibles a travĆ©s del proceso de configuraciĆ³n de acceso unificado de AWS. Ten en cuenta que el ExternalId de este ejemplo incluye un comodĆ­n que coincide con cualquier Lambda Function cuyo nombre comience con function-.

3

AƱade relaciones de confianza a un rol existente en la interfaz de usuario de Atlas

A continuaciĆ³n, debes habilitar tu rol autogestionado de AWS IAM para ejecutar el recurso AWS Lambda.

permission-policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": "arn:aws:lambda:us-east-1:257394458927:function:<function-name>"
        }
    ]
}

  1. Navegue a la pĆ”gina de integraciĆ³n de AWS IAM en su proyecto de Atlas y se debe hacer clic en el botĆ³n Authorize an AWS IAM role.

  2. Crea un nuevo rol (o modifica uno existente) con el role-trust-policy.json que se muestra en el modal.

  3. Una vez que el rol se cree (o el rol existente se actualice con la nueva polĆ­tica de confianza), pega el ARN del rol en la ventana emergente.

  4. En la consola de AWS, ve a IAM > Roles y selecciona tu rol.

  5. En la pestaƱa permissions, aƱade un nuevo "inline permission" para permitir que este rol invoque tus lambda(s). El ejemplo permission-policy.json proporcionado anteriormente agrega el permiso para ejecutar cualquier lambda con el nombre <function-name>.

  6. Por Ćŗltimo, navegue hasta su espacio de trabajo de Atlas Stream Processing, agregue un nuevo AWS Lambda connection y elija el AWS IAM Role ARN que configurĆ³ en el paso anterior.

Conecta tu espacio de trabajo de Atlas Stream Processing a tu funciĆ³n AWS Lambda

Para enviar una solicitud a tu recurso de AWS Lambda desde tu pipeline de Atlas Stream Processing, primero debes agregar tu recurso de AWS Lambda como una conexiĆ³n en tu recurso de Atlas Stream Processing.

Puedes aƱadir tu recurso AWS Lambda como conexiĆ³n a travĆ©s de la interfaz de usuario de Atlas, la Atlas CLI o la API de Atlas, como se muestra en el siguiente ejemplo. Puedes actualizar el marcador de posiciĆ³n roleArn en el ejemplo con el arn de tu configuraciĆ³n de AWS IAM.

Nota

Este comando curl utiliza un token de acceso de cuenta de servicio (OAuth 2.0) para autenticar en lugar de claves API. Para obtener mĆ”s informaciĆ³n, consulte IntroducciĆ³n a la API de administraciĆ³n de Atlas.

curl --header "Authorization: Bearer {ACCESS-TOKEN}" \
   --header "Content-Type: application/json" \
   --header "Accept: application/vnd.atlas.2023-02-01+json" \
   --include \
   --data '{"name": "TestAWSLambdaConnection","type": "AWSLambda","aws": {"roleArn": "arn:aws:iam::<aws_account>:role/<role_name>"}}' \
   --request POST "https://cloud.mongodb.com/api/atlas/v2/groups/<group_id>/streams/<tenant_name>/connections"

Sintaxis

Solicitud mĆ­nima

El siguiente ejemplo muestra los campos obligatorios para una solicitud mĆ­nima.

{ $externalFunction: {
  connectionName: "myLambdaConnection",
  functionName: "arn:aws:lambda:region:account-id:function:function-name",
  as: "response",
}}

Solicitud personalizada

El siguiente ejemplo personalizado especifica el manejo de errores, la ejecuciĆ³n sĆ­ncrona y un documento de Atlas Stream Processing preprocesado como carga Ćŗtil ademĆ”s de los campos requeridos ilustrados anteriormente.

{ $externalFunction: {
        connectionName: "myLambdaConnection",
        functionName: "arn:aws:lambda:region:account-id:function:function-name",
        execution: "sync"
        as: "response",
        onError: "fail",
        payload: [{$replaceRoot: { newRoot: "$fullDocument.payloadToSend" } }, { $addFields: { sum: { $sum: "$randomArray" }}}, { $project: { success: 1, sum: 1 }}],
}}

Nota

El campo onError define el comportamiento para errores a nivel de API tanto para solicitudes sĆ­ncronas como asĆ­ncronas a su recurso AWS Lambda, asĆ­ como errores de la funciĆ³n AWS Lambda para solicitudes sĆ­ncronas.

La etapa $externalFunction procesa un documento con los siguientes campos:

Campo
Tipo
Necesidad
DescripciĆ³n

connectionName

string

Requerido

Etiqueta que identifica la conexiĆ³n en el Registro de Conexiones, a la que se envĆ­a la solicitud.

functionName

string

Requerido

El ARN completo de AWS o el nombre de la funciĆ³n AWS Lambda que se va a activar.

execution

enum

Opcional

ParĆ”metro que especifica si la funciĆ³n de AWS Lambda debe llamarse de forma sĆ­ncrona o asĆ­ncrona. Los valores aceptados son:

  • sync

  • async

Por defecto a sync, que es necesaria si las etapas restantes en su pipeline de Atlas Stream Processing requieren la salida de la funciĆ³n AWS Lambda.

as

string

Condicional

Nombre del campo para la respuesta de la REST API.

Obligatorio cuando execution es "sync" y esta etapa no es la etapa final (receptor) en el pipeline. Al establecer este campo cuando execution es "async" devuelve un error.

Si el endpoint devuelve 0 bytes, el operador no configura el campo as.

onError

string

Opcional

Comportamiento cuando el operador encuentra un HTTPS Status Code o una falla relacionada con Lambda Runtime. Debe ser uno de los siguientes valores:

  • "dlq" : Pase el documento afectado a la fila de letra muerta.

  • "ignore" : No haga nada con el documento afectado.

  • "fail" : Finalizar el procesador de flujo en caso de error.

onError no se activa en errores que surjan de una configuraciĆ³n incorrecta del propio operador $externalFunction, como expresiones no vĆ”lidas.

Se establece por defecto en "dlq".

payload

arreglo

Opcional

Pipeline interno personalizado que permite personalizar el cuerpo de la solicitud enviado al endpoint de la API. payload soporta las siguientes expresiones:

  • $project

  • $addFields

  • $replaceRoot

  • $set

parallelism

entero

Opcional

NĆŗmero mĆ”ximo de solicitudes paralelas realizadas a la funciĆ³n externa. Los valores de paralelismo mĆ”s altos pueden aumentar el rendimiento. Sin embargo, los valores mĆ”s altos tambiĆ©n requieren mĆ”s recursos computacionales.

Para establecer este campo en un valor mayor que 1, debe establecer execution en "sync". Se establece por defecto en 1.

El valor mĆ”ximo que puede establecer depende del nivel de su procesador de transmisiĆ³n. Para obtener mĆ”s informaciĆ³n, consulte GuĆ­a de selecciĆ³n de nivel de Atlas Stream Processing.

Cada procesador de flujo tiene un valor mƔximo de paralelismo acumulado determinado por su nivel. El paralelismo acumulativo de un procesador de flujo se calcula de la siguiente manera:

parallelism total - parallelized stages

Donde parallelism total es la suma de todos los valores parallelism mayores que 1 en las etapas $source, $lookup, $merge, $emit y $externalFunction, y parallelized stages es la cantidad de estas etapas con valores parallelism mayores que 1.

Por ejemplo, si tu etapa $source establece un valor parallelism de 4, tu etapa $lookup no define un valor parallelism (por lo tanto, se toma el valor por defecto 1), y tu etapa $merge define un valor parallelism de 2, entonces tiene dos parallelized stages y el paralelismo acumulativo de su procesador de flujo se calcula como (4 + 2) - 2.

Si un procesador de flujo supera el paralelismo acumulativo mƔximo para su nivel, Atlas Stream Processing genera un error e informa del nivel mƭnimo de procesador requerido para el nivel de paralelismo previsto. Debe escalar el procesador a un nivel superior o reducir los valores de paralelismo de sus etapas para resolver el error. Para aprender mƔs, consulte Procesamiento de flujos.

partitionBy

expresiĆ³n

Opcional

ExpresiĆ³n utilizada para dividir el flujo de entrada en hilos paralelos. Atlas Stream Processing asigna cada documento de entrada con el mismo resultado de expresiĆ³n partitionBy al mismo hilo.

Si este campo no se especifica, Atlas Stream Processing distribuye los documentos de entrada entre los subprocesos paralelos sin tener en cuenta el contenido del documento.

Para establecer este campo, debe establecer execution en "sync" y parallelism en un valor mayor que 1.

Comportamiento

La etapa $externalFunction envĆ­a una solicitud al recurso AWS Lambda especificado. Si la solicitud es sincrĆ³nica, la respuesta se devuelve en el campo especificado y el pipeline continĆŗa procesando. Si la solicitud es asincrĆ³nica, el pipeline continĆŗa procesando sin esperar la respuesta.

Si una solicitud sĆ­ncrona falla, el comportamiento de manejo de errores del pipeline estĆ” determinado por el campo onError. Si la solicitud es asĆ­ncrona, el campo onError se aplica solo a errores de la API de AWS, en lugar de errores de funciĆ³n de AWS Lambda. Si el campo onError no se especifica, el comportamiento por defecto es enviar el documento afectado a la fila de letra muerta.

onError Valor
Comportamiento

"dlq"

El documento afectado se envĆ­a a la fila de letra muerta.

"ignore"

El documento afectado se ignora.

"fail"

El procesador de flujo se termina en caso de error.

Si la canalizaciĆ³n no puede convertir el documento a un JSON correcto o la canalizaciĆ³n configurada no crea un objeto BSON vĆ”lido como producto de la etapa final, el documento se envĆ­a a la fila de letra muerta independientemente del ajuste onError.

Los errores que surgen de una configuraciĆ³n incorrecta del operador $externalFunction en sĆ­, como expresiones no vĆ”lidas, no desencadenan el comportamiento onError. En cambio, el procesador de flujo falla con un mensaje de error. La pipeline de Atlas Stream Processing reintenta las solicitudes fallidas que pueden ser resultado de errores transitorios.

Volver

$merge

Next

$iceberg

En esta pƔgina