Automated Data Enrichment with Lambda and AI Services

Question

Pulumi · Accepted Answer

Automated data enrichment involves taking raw data inputs, processing them through one or more services, typically for the purpose of enhancing, refining, augmenting, or transforming the data. In the context of AWS, this often involves the combination of AWS Lambda, which is a serverless compute service that runs your code in response to events and automatically manages the underlying compute resources for you, and various AI services provided by AWS such as Amazon Comprehend, Amazon Rekognition, etc., which provide pre-trained intelligence for your applications.

In this guide, I'll show you a Pulumi program that automates data enrichment using AWS Lambda and an AWS AI service. Let's choose AWS Comprehend, which provides natural language processing (NLP) capabilities, to analyze text. The process would work as follows:

1. An AWS S3 bucket will be used to store input data.
2. AWS Lambda function will be triggered on the arrival of new data in the S3 bucket.
3. The AWS Lambda function will invoke AWS Comprehend to analyze the text data.
4. The results from AWS Comprehend could be stored back in S3 or another datastore (DynamoDB, for instance).

Here's a Pulumi program in Python that sets this up:

```python
import pulumi
import pulumi_aws as aws

# Create an S3 bucket to store the text files
bucket = aws.s3.Bucket("textFiles")

# IAM role that allows Lambda to call AWS services on your behalf
lambda_role = aws.iam.Role("lambdaRole",
    assume_role_policy="""{
        "Version": "2012-10-17",
        "Statement": [{
            "Action": "sts:AssumeRole",
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            }
        }]
    }"""
)

# Policy granting the necessary permissions for Lambda function
lambda_policy = aws.iam.Policy("lambdaPolicy",
    policy=pulumi.Output.all(bucket.arn).apply(lambda args: f"""{{
        "Version": "2012-10-17",
        "Statement": [
            {{
                "Effect": "Allow",
                "Action": [
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents"
                ],
                "Resource": "arn:aws:logs:*:*:*"
            }},
            {{
                "Effect": "Allow",
                "Action": [
                    "comprehend:DetectEntities",
                    "comprehend:DetectKeyPhrases"
                ],
                "Resource": "*"
            }},
            {{
                "Effect": "Allow",
                "Action": "s3:GetObject",
                "Resource": "{args[0]}/*"
            }}
        ]
    }}"""
    )
)

# Attach the policy to the role
lambda_role_policy_attachment = aws.iam.RolePolicyAttachment("lambdaRolePolicyAttachment",
    role=lambda_role.name,
    policy_arn=lambda_policy.arn
)

# Lambda function that gets invoked when a new text file is uploaded to our S3 bucket
lambda_function = aws.lambda_.Function("textAnalysisFunction",
    code=pulumi.FileArchive("./lambda_code.zip"),
    role=lambda_role.arn,
    handler="lambda_function.handler",
    runtime="python3.8",
    environment={
        "variables": {
            "BUCKET_NAME": bucket.name
        }
    }
)

# S3 bucket event to trigger the Lambda function
s3_event_source = aws.lambda_.EventSourceMapping("s3EventSource",
    batch_size=1,
    event_source_arn=bucket.arn,
    function_name=lambda_function.name,
    starting_position="LATEST"
)

# Set up notification for the S3 bucket to invoke the Lambda function
bucket_notification = aws.s3.BucketNotification("bucketNotification",
    bucket=bucket.id,
    lambda_functions=[{
        "lambda_function_arn": lambda_function.arn,
        "events": ["s3:ObjectCreated:*"],
        "filter_prefix": "inputs/",
        "filter_suffix": ".txt"
    }]
)

# Outputs
pulumi.export('bucket_name', bucket.id)
pulumi.export('lambda_role_arn', lambda_role.arn)
pulumi.export('lambda_function_name', lambda_function.name)
```

In the program:

- We created an S3 bucket where you can upload your text files.
- We defined an IAM role for AWS Lambda with the necessary trust relationship policy.
- We created an IAM policy with permissions to access CloudWatch Logs and call AWS Comprehend on the text data. We also granted the Lambda function read access to the S3 bucket.
- Attached the IAM policy to the IAM role.
- We defined an AWS Lambda function that points to the code (bundled into a .zip file) and defined the runtime, handler, etc., along with environment variables.
- We then created an Event Source Mapping to trigger the Lambda function upon specific S3 bucket events.
- Finally, we used a `BucketNotification` resource to connect the S3 upload events to the Lambda function.

This setup allows for automated processing of text files uploaded to the S3 bucket, extracting insights using AWS Comprehend. The outputs at the end of the program will give you the generated resource names that can be used for reference or further integration.