> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

# `$externalFunction`

## Definition

The [`$externalFunction`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#mongodb-pipeline-pipe.-externalFunction) stage triggers processes in a specific AWS Lambda resource. Your request to the AWS Lambda process can either be synchronous or asynchronous.

## Create an AWS Lambda and Authenticate with Unified AWS Access

In order to call an AWS Lambda resource from within your Atlas Stream Processing pipeline, your AWS Lambda must be deployed to the same AWS region in which your Atlas Stream Processing is deployed. To learn more about deploying an AWS Lambda resource, see the [AWS documentation.](https://docs.aws.amazon.com/lambda/)

1. Create an AWS Lambda Function.

   With either the AWS CLI or through the AWS UI, [create a lambda function.](https://docs.aws.amazon.com/lambda/latest/dg/getting-started.html)

   **Note:**

   - AWS Lambda resources must be created with an `AWS_IAM` authentication type, which requires that you create an AWS IAM [Role](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) and [Policy.](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html)

   - Only the buffered, not streaming, response type is supported.

2. Configure Unified AWS Access.

   **Note:**

   The procedure described here only covers the basic setup flow in the Atlas UI.  To learn more, see the the [Set Up Unified AWS Access](https://www.mongodb.com/docs/atlas/security/set-up-unified-aws-access.md#std-label-set-up-unified-aws-access) documentation.

   **Required Access**

   To set up unified AWS (Amazon Web Services) access, you must have [`Organization Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Organization-Owner) or [`Project Owner`](https://www.mongodb.com/docs/atlas/reference/user-roles.md#mongodb-authrole-Project-Owner) access to the project.

   **Prerequisites**

   - An Atlas [account.](https://www.mongodb.com/docs/atlas/index.md#std-label-atlas-index)

   - The [AWS Command Line Interface (CLI).](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-install.html)

   **Note:**

   - Your AWS IAM policy must include the `lambda:InvokeFunction` and `lambda:GetFunctionConfiguration` actions. Atlas Stream Processing uses `lambda:GetFunctionConfiguration` to retrieve the function timeout so that it waits for the full duration of each execution.

   - You must replace the placeholder `ExternalId` and `Resource` values with your own, which are available through the Unified AWS Access configuration process. Note that the `ExternalId` in this example includes a wildcard that matches any Lambda Function with a name that begins with `function-`.

3. Add Trust Relationships to an Existing Role in the Atlas UI

   Next, you must enable your self-managed AWS IAM role to execute your AWS Lambda resource.

   `permission-policy.json`

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

   ```

   1. Navigate to the AWS IAM integration page in your Atlas project and click on the Authorize an AWS
      IAM role button.

   2. Create a new role (or modify an existing role) with the `role-trust-policy.json` that is shown in the modal.

   3. Once the role is created (or the existing role is updated with the new trust policy), paste the role's ARN in the modal.

   4. In the AWS console, go to IAM > Roles and select your role.

   5. In the permissions tab, add a new "inline permission" to allow this role to invoke your lambda(s). The example `permission-policy.json` provided above adds the permission to run any lambda with the name `<function-name>`.

   6. Finally, navigate to your Atlas Stream Processing Workspace, add a new AWS Lambda connection and choose the AWS IAM Role ARN that you configured in the previous step.

## Connect your Atlas Stream Processing Workspace to your AWS Lambda Function

In order to send a request to your AWS Lambda resource from within your Atlas Stream Processing pipeline, you must first add your AWS Lambda resource as a [connection](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections) in your Atlas Stream Processing resource.

You can add your AWS Lambda resource as a connection through the Atlas UI, the Atlas CLI, or the Atlas API, as shown in the following example. You can update the `roleArn` placeholder in the example with the `arn` from your [AWS IAM configuration.](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns)

**Note:**

This `curl` command uses a service account access token ([OAuth 2.0](https://oauth.net/2/)) to authenticate instead of API keys. To learn more, see [Get Started with the Atlas Administration API.](https://www.mongodb.com/docs/atlas/configure-api-access.md#std-label-atlas-admin-api-access)

```bash
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"
```

## Syntax

### Minimal request

The following example shows the required fields for a minimal request.

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

### Customized request

The following, customized example specifies error handling, synchronous execution and a preprocessed Atlas Stream Processing document as a payload in addition to the required fields illustrated above.

```json
{ $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 }}],
}}
```

**Note:**

The `onError` field defines behavior for API level errors for both synchronous and asynchronous requests to your AWS Lambda resource as well as AWS Lambda function errors for synchronous requests.

The `$externalFunction` stage takes a document with the following fields:

| Field | Type | Necessity | Description |
| --- | --- | --- | --- |
| `connectionName` | string | Required | Label that identifies the connection in the [Connection Registry](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-connection-registry.md#std-label-atlas-sp-manage-connections), to which the request is sent. |
| `functionName` | string | Required | The full AWS ARN or the name of the AWS Lambda function to be triggered. |
| `execution` | enum | Optional | Parameter that specifies whether the AWS Lambda function should be called synchronously or asynchronously. Accepted values are: `sync`; `async` Defaults to `sync`, which is required if remaining stages in your Atlas Stream Processing pipeline require the output from your AWS Lambda function. |
| `as` | string | Conditional | Name of the field for the REST API response. Required when `execution` is `"sync"` and this stage is not the final (sink) stage in the pipeline. Setting this field when `execution` is `"async"` returns an error. If the endpoint returns 0 bytes, the operator doesn't set the `as` field. |
| `onError` | string | Optional | Behavior when the operator encounters an  `HTTPS Status Code` or Lambda Runtime-related failure. Must be one of the following values: `"dlq"` : Pass the affected document to the [dead letter queue.](https://www.mongodb.com/docs/manual/reference/glossary.md#std-term-dead-letter-queue); `"ignore"` : Do nothing with the affected document.; `"fail"` : Terminate the stream processor on error. `onError` does not trigger on errors arising from incorrect configuration of the `$externalFunction` operator itself, such as invalid expressions. Defaults to `"dlq"`. |
| `payload` | array | Optional | Custom inner pipeline that allows you to customize the request body sent to the API endpoint. `payload` supports the following expressions: `$project`; `$addFields`; `$replaceRoot`; `$set` |
| `parallelism` | integer | Optional | Maximum number of parallel requests made to the external function. Higher parallelism values may increase throughput. However, higher values also require more computational resources. To set this field to a value greater than `1`, you must set `execution` to `"sync"`. Defaults to `1`. The maximum value you can set depends on the tier of your stream processor. To learn more, see [Atlas Stream Processing Tier Selection Guide.](https://www.mongodb.com/docs/atlas/atlas-stream-processing/tier-guide.md#std-label-atlas-sp-tier-guide) Each stream processor has a maximum cumulative parallelism value determined by its tier. The cumulative parallelism of a stream processor is calculated as follows: `parallelism total - parallelized stages` Where `parallelism total` is the sum of all `parallelism` values greater than 1 across the [`$source`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-source.md#mongodb-pipeline-pipe.-source), [$lookup](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-lookup.md#std-label-atlas-sp-agg-lookup), [$merge](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-merge.md#std-label-atlas-sp-agg-merge), [$emit](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-emit.md#std-label-atlas-sp-agg-emit), and [$externalFunction](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#std-label-atlas-sp-agg-external-function) stages, and `parallelized stages` is the number of these stages with `parallelism` values greater than `1`. For example, if your `$source` stage sets a `parallelism` value of `4`, your `$lookup` stage sets no `parallelism` value (thus defaulting to `1`), and your `$merge` stage sets a `parallelism` value of `2`, then you have two `parallelized stages`, and the cumulative parallelism of your stream processor is calculated as `(4 + 2) - 2`. If a stream processor exceeds the maximum cumulative parallelism for its tier, Atlas Stream Processing throws an error and advises you of the minimum processor tier required for your intended level of parallelism. You must either scale the processor up to a higher tier or lower the parallelism values of your stages to resolve the error. To learn more, see [Stream Processing.](https://www.mongodb.com/docs/atlas/billing/stream-processing-costs.md#std-label-stream-processing-costs) |
| `partitionBy` | expression | Optional | Expression used to partition the input stream into parallel threads. Atlas Stream Processing allocates each input document with the same `partitionBy` expression result to the same thread. If this field is unspecified, Atlas Stream Processing distributes input documents across the parallel threads without regard to document content. To set this field, you must set `execution` to `"sync"` and `parallelism` to a value greater than `1`. |

## Behavior

The [`$externalFunction`](https://www.mongodb.com/docs/atlas/atlas-stream-processing/sp-agg-externalfunction.md#mongodb-pipeline-pipe.-externalFunction) stage sends a request to the specified AWS Lambda resource. If the request is synchronous, the response is returned in the specified field and the pipeline continues processing. If the request is asynchronous, the pipeline continues processing without waiting for the response.

If a synchronous request fails, the pipeline's error handling behavior is determined by the `onError` field. If the request is asynchronous, the `onError` field only applied to AWS API errors, as opposed to AWS Lambda function errors. If the `onError` field is not specified, the default behavior is to send the affected document to the [dead letter queue.](https://www.mongodb.com/docs/manual/reference/glossary.md#std-term-dead-letter-queue)

| `onError` Value | Behavior |
| --- | --- |
| `"dlq"` | The affected document is sent to the dead letter queue. |
| `"ignore"` | The affected document is ignored. |
| `"fail"` | The stream processor is terminated on error. |

If the pipeline can't convert the document to proper JSON, or the configured pipeline does not create a valid bson object as a product of the final stage, the document is sent to the [dead letter queue](https://www.mongodb.com/docs/manual/reference/glossary.md#std-term-dead-letter-queue) regardless of the `onError` setting.

Errors arising from incorrect configuration of the `$externalFunction` operator itself, such as invalid expressions, do not trigger the `onError` behavior. Instead, the stream processor fails with an error message. The Atlas Stream Processing pipeline retries failed requests that may be the result of transient errors.