Intelligent Document Extraction Pipeline

Build a serverless pipeline that auto-extracts text, tables, and forms from documents.

Introduction

30 Second Summary

Every business deals with stacks of documents. Invoices, tax forms, contracts, and receipts all contain critical data that someone has to manually read and type into a system. That process is slow, error-prone, and expensive.

In this project, you will build an automated document extraction pipeline using AWS serverless services. You'll upload a document to the cloud and watch as your pipeline automatically reads it, pulls out structured data like tables and form fields, and stores the results in a database.

What You'll Build

An event-driven pipeline that automatically processes documents uploaded to S3 by triggering a Lambda function to extract structured data with Amazon Textract and store results in DynamoDB.

By the end of this project, you'll have:

  • An S3 bucket configured as a document intake point with event notifications that automatically trigger processing.
  • A Lambda function that calls Textract to extract tables, forms, and raw text from uploaded documents.
  • A DynamoDB table tracking document metadata, processing status, confidence scores, and extraction results.
  • Secret Mission: Add natural-language queries to ask targeted questions about your documents and get specific answers.

Are there any prerequisites?

You need an AWS account that is less than 3 months old to access the Textract free tier. You also need the AWS CLI installed and configured, and Python installed on your machine.

Set Up Your AWS Infrastructure

Before your pipeline can process documents automatically, it needs three foundational AWS resources working together.

In this step, you'll create an S3 bucket to receive uploads, a DynamoDB table to store extraction results, and an IAM role that gives your future Lambda function the permissions it needs to connect everything.

In this step, get ready to:
  • Create an S3 bucket for document uploads.
  • Create a DynamoDB table to store document metadata.
  • Create an IAM role with least-privilege permissions for the Lambda function.
Create an S3 bucket for document uploads

Your pipeline needs an intake point where documents land before processing. An S3 bucket acts as this intake point.

  • Open your terminal.
  • Create your S3 bucket by running this command:
aws s3 mb s3://[[BUCKET_NAME="doc-engine-your-initials-todays-date"]]

What does this command do?

The aws s3 mb command creates ("makes") a new S3 bucket.

S3 bucket names must be globally unique across all of AWS. Using your initials and today's date (e.g., doc-engine-jd-20260704) helps ensure uniqueness. Only use lowercase letters, numbers, and hyphens.

  • Verify the bucket was created by running:
aws s3 ls
  • You should see your new bucket name in the list of S3 buckets.

Don't see your bucket?

  • Check that your bucket name contains only lowercase letters, numbers, and hyphens. No uppercase letters or underscores.
  • If you see a "BucketAlreadyExists" error, your bucket name is taken. Try adding more characters to make it unique.
  • Make sure your AWS CLI is configured with valid credentials.

Why isn't my S3 bucket being created?

Create a DynamoDB table for document metadata

Your pipeline needs somewhere to store the results after Textract extracts data from each document. A DynamoDB table will track each document's processing status, extracted text, and form data.

  • Create the DocumentMetadata table by running this command:
aws dynamodb create-table \
    --table-name DocumentMetadata \
    --attribute-definitions AttributeName=document_id,AttributeType=S \
    --key-schema AttributeName=document_id,KeyType=HASH \
    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5

What does this command do?

  • --table-name DocumentMetadata names the table where all extraction results will be stored.
  • --attribute-definitions defines document_id as a String attribute. This will be the unique identifier for each document.
  • --key-schema with KeyType=HASH makes document_id the partition key (the primary way DynamoDB finds records).
  • --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 sets the table's read and write capacity. These values stay well within the DynamoDB always-free tier of 25 RCU and 25 WCU.

Why provisioned mode instead of on-demand?

DynamoDB's always-free tier (25 GB storage, 25 RCU, 25 WCU) only applies to provisioned capacity mode. On-demand mode does not qualify for the free tier, so you'd be charged for every read and write.

  • Verify the table was created by running:
aws dynamodb describe-table --table-name DocumentMetadata
  • You should see a JSON response with "TableStatus": "ACTIVE" (or "CREATING" if it's still provisioning). The table will be ready within a few seconds.

Seeing a ResourceInUseException?

This means a table named DocumentMetadata already exists in your account's default region. You can delete it first with aws dynamodb delete-table --table-name DocumentMetadata and wait a moment before recreating it.

Help me fix my DynamoDB table creation error.

Create an IAM role for the Lambda function

Your Lambda function needs permission to read from S3, call Textract, and write to DynamoDB. An IAM role bundles these permissions together so Lambda can assume them at runtime.

First, you'll create the role with a trust policy. Then you'll attach an inline permissions policy.

  • Create a file called trust-policy.json with the following content:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

What is a trust policy?

A trust policy defines who is allowed to "assume" (use) this role. By specifying lambda.amazonaws.com as the principal, you're saying only the AWS Lambda service can take on these permissions.

  • Create the IAM role by running this command:
aws iam create-role \
    --role-name DocEngineRole \
    --assume-role-policy-document file://trust-policy.json
  • You should see a JSON response containing the role's ARN (Amazon Resource Name). It will look something like arn:aws:iam::123456789012:role/DocEngineRole.
  • Now create the permissions policy. Create a file called permissions-policy.json with the following content:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "textract:AnalyzeDocument",
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::[[BUCKET_NAME="doc-engine-your-initials-todays-date"]]/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:PutItem",
        "dynamodb:UpdateItem"
      ],
      "Resource": "arn:aws:dynamodb:*:*:table/DocumentMetadata"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "*"
    }
  ]
}

What does this policy grant?

  • textract:AnalyzeDocument lets Lambda call Textract to extract text, tables, and forms from documents.
  • s3:GetObject lets Lambda read the uploaded document from your bucket.
  • dynamodb:PutItem and dynamodb:UpdateItem let Lambda write and update records in the DocumentMetadata table.
  • The logs:* permissions let Lambda write execution logs to CloudWatch for debugging.

This follows the principle of least privilege. Each permission targets only the specific actions and resources the function needs.

  • Attach the inline policy to the role by running:
aws iam put-role-policy \
    --role-name DocEngineRole \
    --policy-name DocEnginePolicy \
    --policy-document file://permissions-policy.json
  • Verify the role exists by running:
aws iam get-role --role-name DocEngineRole
  • You should see a JSON response showing the role details including its ARN and the trust policy you created.

Seeing an EntityAlreadyExists error?

This means a role named DocEngineRole already exists in your account. You can delete it with aws iam delete-role-policy --role-name DocEngineRole --policy-name DocEnginePolicy followed by aws iam delete-role --role-name DocEngineRole, then re-run the creation commands.

Help me fix my IAM role creation error.

Your foundational infrastructure is in place. Next up, you'll build the Lambda function that ties these resources together to process documents with Textract.

Build the Lambda Function

Your infrastructure is ready. Now you need the code that actually processes documents when they land in your S3 bucket.

In this step, you will create an AWS Lambda function that receives S3 upload events, calls Amazon Textract to extract structured data from documents, and stores the results in DynamoDB. The Lambda runtime already includes the boto3 SDK, so no external packages are needed.

In this step, get ready to:
  • Create a Lambda function with the correct runtime, role, and resource settings.
  • Write handler code that calls Textract and parses extracted data.
  • Deploy and test the function with a sample S3 event.
Create the Lambda function in the AWS Console

Lambda functions need a runtime (the language version), an execution role (what AWS services the function can access), and resource limits tuned to the workload.

  • In the AWS Console, search for Lambda in the top search bar and select it.
  • Click Create function.
  • Select Author from scratch.
  • For Function name, enter DocEngineProcessor.
  • For Runtime, choose Python 3.13.
  • Expand Change default execution role, select Use an existing role, and choose DocEngineRole from the dropdown.
  • Click Create function.

Now configure the function's resource limits and environment variable.

  • On the function page, click the Configuration tab.
  • Select General configuration in the left sidebar, then click Edit.
  • Set Memory to 256 MB.
  • Set Timeout to 1 min 0 sec (60 seconds total).
  • Click Save.

Why 256 MB and 60 seconds?

Textract responses for documents with tables and forms can be several megabytes of JSON. The extra memory gives your function room to parse those responses without running out of memory.

Textract synchronous calls typically take 5 to 15 seconds depending on document complexity. The default 3-second timeout would cause failures before Textract even responds.

  • Still in the Configuration tab, select Environment variables in the left sidebar.
  • Click Edit, then Add environment variable.
  • For Key, enter TABLE_NAME.
  • For Value, enter DocumentMetadata.
  • Click Save.

Why use an environment variable for the table name?

Storing the table name in an environment variable keeps your code portable. If you ever rename the table or deploy to a different environment, you change the variable instead of editing the code.

Write the handler code

The Lambda console created a default lambda_function.py file with placeholder code. You will replace it with the document processing logic.

  • Click the Code tab to open the inline code editor.
  • Select all the existing code in lambda_function.py and delete it.
  • Paste in the following imports and setup code:
import json
import boto3
import os
import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)

textract_client = boto3.client('textract')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ.get('TABLE_NAME', 'DocumentMetadata'))

What does this setup do?

  • Creates a textract_client for calling the Textract API and a dynamodb resource for writing to your table.
  • Reads TABLE_NAME from the environment variable you just configured.
  • Sets up logging so you can debug via CloudWatch Logs.
  • These clients are created outside the handler so they persist across invocations in a warm Lambda environment.
  • Below the setup code, paste the main handler function:
def lambda_handler(event, context):
    logger.info(f"Received event: {json.dumps(event)}")

    record = event['Records'][0]
    bucket = record['s3']['bucket']['name']
    document_key = record['s3']['object']['key']
    document_id = document_key

    # Update status to PROCESSING
    table.put_item(
        Item={
            'document_id': document_id,
            'bucket': bucket,
            'upload_timestamp': datetime.now(timezone.utc).isoformat(),
            'processing_status': 'PROCESSING'
        }
    )

    try:
        # Call Textract AnalyzeDocument
        response = textract_client.analyze_document(
            Document={
                'S3Object': {
                    'Bucket': bucket,
                    'Name': document_key
                }
            },
            FeatureTypes=['TABLES', 'FORMS']
        )

        blocks = response['Blocks']

        # Extract raw text
        raw_text = extract_raw_text(blocks)

        # Extract key-value pairs from forms
        key_values = extract_key_values(blocks)

        # Filter low confidence fields
        high_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] >= 80}
        low_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] < 80}

        # Extract tables
        tables = extract_tables(blocks)

        # Count pages
        page_count = sum(1 for b in blocks if b['BlockType'] == 'PAGE')

        # Update DynamoDB with results
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, processed_at = :ts, '
                           'raw_text = :text, extracted_key_values = :kvs, '
                           'low_confidence_fields = :low_kvs, '
                           'tables = :tables, table_count = :tc, page_count = :pc',
            ExpressionAttributeValues={
                ':status': 'COMPLETED',
                ':ts': datetime.now(timezone.utc).isoformat(),
                ':text': raw_text[:400000],  # DynamoDB item size limit
                ':kvs': high_confidence_kvs,
                ':low_kvs': low_confidence_kvs,
                ':tables': tables,
                ':tc': len(tables),
                ':pc': page_count
            }
        )

        logger.info(f"Successfully processed {document_id}: {len(high_confidence_kvs)} key-value pairs, {len(tables)} tables")
        return {'statusCode': 200, 'body': f'Processed {document_id}'}

    except Exception as e:
        logger.error(f"Error processing {document_id}: {str(e)}")
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, error_message = :err, processed_at = :ts',
            ExpressionAttributeValues={
                ':status': 'FAILED',
                ':err': str(e),
                ':ts': datetime.now(timezone.utc).isoformat()
            }
        )
        raise

What does the handler do?

  • Extracts the bucket name and object key from the S3 event payload.
  • Immediately writes a PROCESSING status to DynamoDB so you can track in-flight documents.
  • Calls textract_client.analyze_document() with FeatureTypes=['TABLES', 'FORMS'] to extract structured data from the document.
  • Parses the response into raw text, key-value pairs (form fields), and table data using helper functions.
  • Filters key-value pairs by confidence score. Anything below 80% goes into a separate low_confidence_fields attribute.
  • Updates DynamoDB with all results on success, or writes a FAILED status with the error message if anything goes wrong.
  • Below the handler, paste the helper functions that parse the Textract response:
def extract_raw_text(blocks):
    lines = [b['Text'] for b in blocks if b['BlockType'] == 'LINE']
    return '\n'.join(lines)


def extract_key_values(blocks):
    block_map = {b['Id']: b for b in blocks}
    key_values = {}

    for block in blocks:
        if block['BlockType'] == 'KEY_VALUE_SET' and 'KEY' in block.get('EntityTypes', []):
            key_text = get_text_from_block(block, block_map)
            value_block = get_value_block(block, block_map)
            if value_block:
                value_text = get_text_from_block(value_block, block_map)
                confidence = block.get('Confidence', 0)
                if key_text.strip():
                    key_values[key_text.strip()] = {
                        'value': value_text.strip(),
                        'confidence': round(confidence, 2)
                    }

    return key_values


def get_value_block(key_block, block_map):
    for rel in key_block.get('Relationships', []):
        if rel['Type'] == 'VALUE':
            for value_id in rel['Ids']:
                return block_map.get(value_id)
    return None


def get_text_from_block(block, block_map):
    text = ''
    for rel in block.get('Relationships', []):
        if rel['Type'] == 'CHILD':
            for child_id in rel['Ids']:
                child = block_map.get(child_id)
                if child and child['BlockType'] == 'WORD':
                    text += child.get('Text', '') + ' '
    return text


def extract_tables(blocks):
    block_map = {b['Id']: b for b in blocks}
    tables = []

    for block in blocks:
        if block['BlockType'] == 'TABLE':
            table_data = {}
            for rel in block.get('Relationships', []):
                if rel['Type'] == 'CHILD':
                    for cell_id in rel['Ids']:
                        cell = block_map.get(cell_id)
                        if cell and cell['BlockType'] == 'CELL':
                            row_idx = cell.get('RowIndex', 0)
                            col_idx = cell.get('ColumnIndex', 0)
                            cell_text = get_text_from_block(cell, block_map).strip()
                            if row_idx not in table_data:
                                table_data[row_idx] = {}
                            table_data[row_idx][col_idx] = cell_text

            # Convert to list of rows
            if table_data:
                max_row = max(table_data.keys())
                max_col = max(col for row in table_data.values() for col in row.keys())
                structured_table = []
                for r in range(1, max_row + 1):
                    row = []
                    for c in range(1, max_col + 1):
                        row.append(table_data.get(r, {}).get(c, ''))
                    structured_table.append(row)
                tables.append(structured_table)

    return tables

What do these helper functions do?

  • extract_raw_text collects all LINE blocks and joins them into plain text.
  • extract_key_values finds KEY_VALUE_SET blocks (form fields like "Name: John") and pairs each key with its value and confidence score.
  • get_value_block follows the VALUE relationship from a key block to find its corresponding value block.
  • get_text_from_block follows CHILD relationships to extract the actual text (WORD blocks) from any parent block.
  • extract_tables reconstructs tables by following TABLE to CELL relationships, using RowIndex and ColumnIndex to build a list of rows.

How Textract Blocks link together

Textract returns a flat list of Block objects. Each block has an Id and uses Relationships to point to other blocks. A KEY_VALUE_SET block with EntityType KEY links to a VALUE block via a VALUE relationship. TABLE blocks link to CELL blocks via CHILD relationships.

Your helper functions navigate these relationships to reconstruct the document's structure from the flat block list.

✔️ Awesome, I've got everything!

Great. Make sure you have all three sections (imports/setup, handler, and helpers) pasted in order in lambda_function.py.

ⓧ I'd like to double check the full code

Here is the complete lambda_function.py file. Make sure your code matches this exactly.

import json
import boto3
import os
import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)

textract_client = boto3.client('textract')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ.get('TABLE_NAME', 'DocumentMetadata'))


def lambda_handler(event, context):
    logger.info(f"Received event: {json.dumps(event)}")

    record = event['Records'][0]
    bucket = record['s3']['bucket']['name']
    document_key = record['s3']['object']['key']
    document_id = document_key

    # Update status to PROCESSING
    table.put_item(
        Item={
            'document_id': document_id,
            'bucket': bucket,
            'upload_timestamp': datetime.now(timezone.utc).isoformat(),
            'processing_status': 'PROCESSING'
        }
    )

    try:
        # Call Textract AnalyzeDocument
        response = textract_client.analyze_document(
            Document={
                'S3Object': {
                    'Bucket': bucket,
                    'Name': document_key
                }
            },
            FeatureTypes=['TABLES', 'FORMS']
        )

        blocks = response['Blocks']

        # Extract raw text
        raw_text = extract_raw_text(blocks)

        # Extract key-value pairs from forms
        key_values = extract_key_values(blocks)

        # Filter low confidence fields
        high_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] >= 80}
        low_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] < 80}

        # Extract tables
        tables = extract_tables(blocks)

        # Count pages
        page_count = sum(1 for b in blocks if b['BlockType'] == 'PAGE')

        # Update DynamoDB with results
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, processed_at = :ts, '
                           'raw_text = :text, extracted_key_values = :kvs, '
                           'low_confidence_fields = :low_kvs, '
                           'tables = :tables, table_count = :tc, page_count = :pc',
            ExpressionAttributeValues={
                ':status': 'COMPLETED',
                ':ts': datetime.now(timezone.utc).isoformat(),
                ':text': raw_text[:400000],  # DynamoDB item size limit
                ':kvs': high_confidence_kvs,
                ':low_kvs': low_confidence_kvs,
                ':tables': tables,
                ':tc': len(tables),
                ':pc': page_count
            }
        )

        logger.info(f"Successfully processed {document_id}: {len(high_confidence_kvs)} key-value pairs, {len(tables)} tables")
        return {'statusCode': 200, 'body': f'Processed {document_id}'}

    except Exception as e:
        logger.error(f"Error processing {document_id}: {str(e)}")
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, error_message = :err, processed_at = :ts',
            ExpressionAttributeValues={
                ':status': 'FAILED',
                ':err': str(e),
                ':ts': datetime.now(timezone.utc).isoformat()
            }
        )
        raise


def extract_raw_text(blocks):
    lines = [b['Text'] for b in blocks if b['BlockType'] == 'LINE']
    return '\n'.join(lines)


def extract_key_values(blocks):
    block_map = {b['Id']: b for b in blocks}
    key_values = {}

    for block in blocks:
        if block['BlockType'] == 'KEY_VALUE_SET' and 'KEY' in block.get('EntityTypes', []):
            key_text = get_text_from_block(block, block_map)
            value_block = get_value_block(block, block_map)
            if value_block:
                value_text = get_text_from_block(value_block, block_map)
                confidence = block.get('Confidence', 0)
                if key_text.strip():
                    key_values[key_text.strip()] = {
                        'value': value_text.strip(),
                        'confidence': round(confidence, 2)
                    }

    return key_values


def get_value_block(key_block, block_map):
    for rel in key_block.get('Relationships', []):
        if rel['Type'] == 'VALUE':
            for value_id in rel['Ids']:
                return block_map.get(value_id)
    return None


def get_text_from_block(block, block_map):
    text = ''
    for rel in block.get('Relationships', []):
        if rel['Type'] == 'CHILD':
            for child_id in rel['Ids']:
                child = block_map.get(child_id)
                if child and child['BlockType'] == 'WORD':
                    text += child.get('Text', '') + ' '
    return text


def extract_tables(blocks):
    block_map = {b['Id']: b for b in blocks}
    tables = []

    for block in blocks:
        if block['BlockType'] == 'TABLE':
            table_data = {}
            for rel in block.get('Relationships', []):
                if rel['Type'] == 'CHILD':
                    for cell_id in rel['Ids']:
                        cell = block_map.get(cell_id)
                        if cell and cell['BlockType'] == 'CELL':
                            row_idx = cell.get('RowIndex', 0)
                            col_idx = cell.get('ColumnIndex', 0)
                            cell_text = get_text_from_block(cell, block_map).strip()
                            if row_idx not in table_data:
                                table_data[row_idx] = {}
                            table_data[row_idx][col_idx] = cell_text

            # Convert to list of rows
            if table_data:
                max_row = max(table_data.keys())
                max_col = max(col for row in table_data.values() for col in row.keys())
                structured_table = []
                for r in range(1, max_row + 1):
                    row = []
                    for c in range(1, max_col + 1):
                        row.append(table_data.get(r, {}).get(c, ''))
                    structured_table.append(row)
                tables.append(structured_table)

    return tables
Deploy and test the function

Before wiring up the S3 trigger, you should confirm the function deploys without syntax errors and that the error handling works.

  • In the code editor, click Deploy to save and deploy your function code.
  • Click the Test tab.
  • Under Test event action, select Create new event.
  • For Event name, enter TestS3Event.
  • For Template, select S3 Put.
  • Replace the Event JSON with the following, substituting your bucket name:
{
  "Records": [
    {
      "eventVersion": "2.0",
      "eventSource": "aws:s3",
      "awsRegion": "us-east-1",
      "eventTime": "1970-01-01T00:00:00.000Z",
      "eventName": "ObjectCreated:Put",
      "userIdentity": {
        "principalId": "EXAMPLE"
      },
      "requestParameters": {
        "sourceIPAddress": "127.0.0.1"
      },
      "responseElements": {
        "x-amz-request-id": "EXAMPLE123456789",
        "x-amz-id-2": "EXAMPLE123/5678abcdefghijklambdaisawesome/mnopqrstuvwxyzABCDEFGH"
      },
      "s3": {
        "s3SchemaVersion": "1.0",
        "configurationId": "testConfigRule",
        "bucket": {
          "name": "[[BUCKET_NAME=\"your-bucket-name\"]]",
          "ownerIdentity": {
            "principalId": "EXAMPLE"
          },
          "arn": "arn:aws:s3:::[[BUCKET_NAME=\"your-bucket-name\"]]"
        },
        "object": {
          "key": "uploads/sample-doc.pdf",
          "size": 1024,
          "eTag": "0123456789abcdef0123456789abcdef",
          "sequencer": "0A1B2C3D4E5F678901"
        }
      }
    }
  ]
}
  • Replace us-east-1 with the region where you created your bucket.
  • Click Save, then click Test.

The function will attempt to call Textract on a file that does not exist yet. This is expected to produce a controlled failure.

  • Expand Details under Execution result. You should see an error message indicating that Textract could not find the object in S3.

Why is an error expected here?

The test event references a file (uploads/sample-doc.pdf) that does not exist in your bucket yet. Textract cannot read a non-existent file, so it throws an error.

This is actually a successful test. It confirms three things: your function deploys without syntax errors, the error handling catches the exception, and DynamoDB receives a FAILED status record.

  • Verify the DynamoDB write worked by running this in your terminal:
aws dynamodb get-item --table-name DocumentMetadata --key '{"document_id": {"S": "uploads/sample-doc.pdf"}}'
  • You should see a record with processing_status set to FAILED and an error_message attribute. This confirms your function's error handling and DynamoDB integration work correctly.

Seeing an AccessDeniedException?

  • If the error mentions Textract access denied, confirm that your DocEngineRole inline policy includes textract:AnalyzeDocument on Resource: "*".
  • If DynamoDB returns access denied, check that the policy includes dynamodb:PutItem and dynamodb:UpdateItem on your table's ARN.
  • Make sure you selected DocEngineRole (not the default Lambda role) when you created the function.

Help me troubleshoot my Lambda function permissions.

Your Lambda function is deployed and the error handling is confirmed. In the next step, you will wire up the S3 trigger so documents automatically flow through this pipeline when uploaded.

Connect S3 to Lambda with Event Notifications

Your Lambda function can extract data from documents, but right now it only runs when you manually invoke it with a test event. In a real-world pipeline, you want processing to happen automatically the moment a document arrives.

In this step, you will wire up an S3 event notification that triggers your Lambda function whenever a new file lands in the bucket. Then you will upload a real document and verify the entire pipeline end-to-end.

In this step, get ready to:
  • Add an S3 trigger to your Lambda function with prefix and suffix filters.
  • Upload a sample document to the S3 bucket.
  • Verify the end-to-end flow in CloudWatch Logs and DynamoDB.
Add an S3 trigger to the Lambda function

S3 event notifications let a bucket push events to other AWS services whenever objects are created or deleted. You will configure the trigger from the Lambda console, which automatically grants S3 permission to invoke the function.

  • Return to the Lambda console and open your DocEngineProcessor function.
  • In the Function overview pane, click Add trigger.
  • Select S3 from the trigger type dropdown.
  • Under Bucket, select your S3 bucket (e.g., doc-engine-<initials>-<date>).
  • Under Event types, confirm that All object create events is selected.
  • Enter uploads/ in the Prefix field.
  • Enter .pdf in the Suffix field.
  • Under Recursive invocation, select the checkbox to acknowledge the warning about using the same bucket for input and output.
  • Click Add.

What does this trigger configuration do?

The prefix filter uploads/ ensures only files uploaded to the uploads/ folder trigger the function. The suffix filter .pdf further limits triggers to PDF files only.

This prevents the function from being invoked by unrelated objects in the bucket (like log files or temporary data). It also prevents recursive invocations, since the function writes results to DynamoDB rather than back into the same bucket path.

Why acknowledge the recursive invocation warning?

AWS warns you because if a Lambda function writes back to the same bucket that triggers it, the function could invoke itself in an infinite loop. Your function stores results in DynamoDB, not S3, so recursion is not a risk here. The checkbox simply confirms you understand the pattern.

You should see S3 appear as a trigger in the Function overview diagram.

What about the propagation delay?

Changes to bucket notification configurations can take about 5 minutes to take effect. Wait a few minutes before uploading your test document to ensure the trigger is active.

Trigger creation failed?

If you see an error when adding the trigger, check that your IAM role DocEngineRole has the s3:GetObject permission on your bucket. Also confirm that your Lambda function and S3 bucket are in the same AWS Region.

help me troubleshoot why my S3 trigger failed to create for my Lambda function

Upload a sample document to S3

To test the pipeline, you need a document with structured content that Textract can extract. A PDF with tables and form fields works best. You can use any sample invoice, tax form, or structured document you have on hand.

  • Find or download a sample PDF document that contains tables or form fields (an invoice or receipt works well).
  • Upload the document to your S3 bucket's uploads/ folder by running this command:
aws s3 cp [[SAMPLE_DOC_PATH="path/to/your/sample-doc.pdf"]] s3://[[BUCKET_NAME="your-bucket-name"]]/uploads/sample-doc.pdf

What does this command do?

The aws s3 cp command copies a local file to your S3 bucket. By placing the file under the uploads/ prefix with a .pdf suffix, it matches both the prefix and suffix filters you configured on the trigger.

  • Confirm the file was uploaded by running:
aws s3 ls s3://[[BUCKET_NAME="your-bucket-name"]]/uploads/

You should see sample-doc.pdf listed in the output.

Textract supported formats

Amazon Textract supports JPEG, PNG, PDF, and TIFF formats. For this project, single-page documents work best because the AnalyzeDocument API is synchronous and limited to single-page inputs.

Upload command failing?

Make sure you replace the file path with the actual path to your local PDF file. On macOS, you can drag the file into the terminal to auto-fill its path.

help me fix an aws s3 cp upload failure

Verify the end-to-end pipeline

If everything is wired correctly, your upload should have triggered the Lambda function automatically. Now you will confirm the execution succeeded and the results landed in DynamoDB.

  • Open the CloudWatch console.
  • In the left navigation, click Logs, then Log groups.
  • Click the log group named /aws/lambda/DocEngineProcessor.
  • Open the most recent log stream.
  • Look for a log entry containing Successfully processed followed by the count of key-value pairs and tables extracted.

What should I see in the logs?

A successful execution shows the Received event log (the raw S3 event), followed by the Successfully processed message. If you see an error instead, the log will show the exception details to help you debug.

  • Now query DynamoDB to confirm the metadata record was created by running this command:
aws dynamodb get-item --table-name DocumentMetadata --key '{"document_id": {"S": "uploads/sample-doc.pdf"}}'

What does this command do?

This retrieves the specific item from your DocumentMetadata table using the partition key document_id set to uploads/sample-doc.pdf. The Lambda function uses the S3 object key as the document ID, so it matches the file path in the bucket.

You should see a JSON response containing the full DynamoDB item with processing_status set to COMPLETED, along with extracted_key_values, table_count, raw_text, page_count, and processed_at populated.

DynamoDB item shows FAILED status?

Check the error_message field in the DynamoDB item for details. Common issues include the document being multi-page (use a single-page PDF) or the IAM role missing the textract:AnalyzeDocument permission.

help me debug why my Textract document processing failed

No log stream or DynamoDB item at all?

If no log stream appeared, the trigger may not have fired. Wait 5 minutes for the event notification to propagate, then re-upload the file. Also double-check that your file path starts with uploads/ and ends with .pdf to match the filters.

help me troubleshoot why my S3 trigger is not invoking my Lambda function

Your event-driven pipeline is fully operational. Next up, you will enhance the extraction logic to produce more structured and production-quality output.

Enhance the Extraction Output

Your pipeline processes documents end-to-end. But a raw extraction dump is only useful if you can trust it and read it.

In this step, you'll walk through three production-quality features already in your Lambda code: structured table reconstruction, confidence-based filtering, and proper error handling with status tracking.

In this step, get ready to:
  • Review how tables are reconstructed as structured nested lists.
  • Review how confidence scoring filters unreliable key-value extractions.
  • Verify error handling and status transitions by uploading a new document.
Review the structured table extraction

Amazon Textract returns table data as a flat list of TABLE and CELL blocks linked by relationships. That's difficult to use directly. Your extract_tables function reconstructs tables into a readable format: a list of rows where each row is a list of cell text values.

  • Return to the Lambda console and open your DocEngineProcessor function.
  • Scroll down to the extract_tables function in the code editor. Find this section:
def extract_tables(blocks):
    block_map = {b['Id']: b for b in blocks}
    tables = []

    for block in blocks:
        if block['BlockType'] == 'TABLE':
            table_data = {}
            for rel in block.get('Relationships', []):
                if rel['Type'] == 'CHILD':
                    for cell_id in rel['Ids']:
                        cell = block_map.get(cell_id)
                        if cell and cell['BlockType'] == 'CELL':
                            row_idx = cell.get('RowIndex', 0)
                            col_idx = cell.get('ColumnIndex', 0)
                            cell_text = get_text_from_block(cell, block_map).strip()
                            if row_idx not in table_data:
                                table_data[row_idx] = {}
                            table_data[row_idx][col_idx] = cell_text

            # Convert to list of rows
            if table_data:
                max_row = max(table_data.keys())
                max_col = max(col for row in table_data.values() for col in row.keys())
                structured_table = []
                for r in range(1, max_row + 1):
                    row = []
                    for c in range(1, max_col + 1):
                        row.append(table_data.get(r, {}).get(c, ''))
                    structured_table.append(row)
                tables.append(structured_table)

    return tables

How does table reconstruction work?

Textract's response links TABLE blocks to CELL blocks via CHILD relationships. Each CELL has a RowIndex and ColumnIndex telling you where it belongs in the grid.

The function collects all cells into a dictionary keyed by row and column, then converts that dictionary into a list of lists. The result is a clean nested structure like [['Name', 'Amount'], ['Item A', '$50']] that's easy to read and query in DynamoDB.

Review the confidence scoring logic

Amazon Textract assigns a confidence score (0-100) to every extraction. Low-confidence results can introduce bad data into downstream systems. Your function separates trusted extractions from unreliable ones.

  • Scroll up to the lambda_handler function and find the confidence filtering section:
        # Filter low confidence fields
        high_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] >= 80}
        low_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] < 80}

Why 80% as the threshold?

The 80% threshold is a practical starting point. Fields scoring below 80% are stored separately in low_confidence_fields so you can review them manually rather than discarding them entirely.

The extract_key_values function attaches the confidence score to every key-value pair it returns. This lets the filtering logic work without a second pass through the Textract response.

Review error handling and status tracking

Production pipelines need observable state. Your function writes a PROCESSING status before calling Amazon Textract, then updates to COMPLETED or FAILED depending on the outcome.

  • Find the error handling block near the bottom of lambda_handler:
    except Exception as e:
        logger.error(f"Error processing {document_id}: {str(e)}")
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, error_message = :err, processed_at = :ts',
            ExpressionAttributeValues={
                ':status': 'FAILED',
                ':err': str(e),
                ':ts': datetime.now(timezone.utc).isoformat()
            }
        )
        raise

What does the status tracking give you?

The put_item at the start writes PROCESSING immediately. If the function crashes without reaching either update, the record stays in PROCESSING state. That makes stuck documents easy to find.

The processed_at timestamp and page_count give you observability into how long processing took and how large the document was. The error_message field captures the exact failure reason for debugging.

Why does the function re-raise the exception?

After logging the error and writing FAILED to DynamoDB, the raise statement propagates the exception to Lambda. This ensures CloudWatch marks the invocation as failed, which triggers Lambda's built-in retry and dead-letter-queue behavior if configured.

Verify enhanced output with a new document

Now that you understand how each enhancement works, upload a new document to see the structured results in DynamoDB.

  • Upload a new document (a different PDF or image with tables and form fields) to your S3 bucket by running:
aws s3 cp [[DOCPATH="path/to/your-new-document.pdf"]] s3://[[BUCKET="your-bucket-name"]]/uploads/new-document.pdf

Pick a document with tables

Invoices, tax forms, or receipts with line items work well. The more structured data your document has, the more interesting the extraction output will be. Textract supports JPEG, PNG, PDF, and TIFF formats.

  • Wait about 10 seconds for Lambda to process the document, then query DynamoDB to see the results by running:
aws dynamodb get-item --table-name DocumentMetadata --key '{"document_id": {"S": "uploads/new-document.pdf"}}'
  • You should see a DynamoDB item with processing_status set to COMPLETED.
  • Verify the tables attribute contains nested lists (rows of cell values) rather than raw block data.
  • Check that extracted_key_values entries include a confidence score alongside each value.
  • Confirm processed_at, page_count, and table_count are all populated.

Seeing PROCESSING instead of COMPLETED?

The function may still be running. Wait 30 seconds and query again. If the status stays at PROCESSING, check CloudWatch Logs for your Lambda function to see if it timed out or hit an error.

If you see a FAILED status, check the error_message field in the DynamoDB item for details. Common causes include unsupported file formats or documents larger than what the synchronous API supports.

Help me debug my Lambda function's FAILED status in CloudWatch Logs.

✔️ Awesome, I've got everything!

Great. Your enhanced pipeline produces structured, confidence-scored results with full status tracking.

ⓧ I'd like to double check the full code

Here is the complete lambda_function.py as it should look in your Lambda console:

import json
import boto3
import os
import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)

textract_client = boto3.client('textract')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ.get('TABLE_NAME', 'DocumentMetadata'))


def lambda_handler(event, context):
    logger.info(f"Received event: {json.dumps(event)}")

    record = event['Records'][0]
    bucket = record['s3']['bucket']['name']
    document_key = record['s3']['object']['key']
    document_id = document_key

    # Update status to PROCESSING
    table.put_item(
        Item={
            'document_id': document_id,
            'bucket': bucket,
            'upload_timestamp': datetime.now(timezone.utc).isoformat(),
            'processing_status': 'PROCESSING'
        }
    )

    try:
        # Call Textract AnalyzeDocument
        response = textract_client.analyze_document(
            Document={
                'S3Object': {
                    'Bucket': bucket,
                    'Name': document_key
                }
            },
            FeatureTypes=['TABLES', 'FORMS']
        )

        blocks = response['Blocks']

        # Extract raw text
        raw_text = extract_raw_text(blocks)

        # Extract key-value pairs from forms
        key_values = extract_key_values(blocks)

        # Filter low confidence fields
        high_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] >= 80}
        low_confidence_kvs = {k: v for k, v in key_values.items() if v['confidence'] < 80}

        # Extract tables
        tables = extract_tables(blocks)

        # Count pages
        page_count = sum(1 for b in blocks if b['BlockType'] == 'PAGE')

        # Update DynamoDB with results
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, processed_at = :ts, '
                           'raw_text = :text, extracted_key_values = :kvs, '
                           'low_confidence_fields = :low_kvs, '
                           'tables = :tables, table_count = :tc, page_count = :pc',
            ExpressionAttributeValues={
                ':status': 'COMPLETED',
                ':ts': datetime.now(timezone.utc).isoformat(),
                ':text': raw_text[:400000],  # DynamoDB item size limit
                ':kvs': high_confidence_kvs,
                ':low_kvs': low_confidence_kvs,
                ':tables': tables,
                ':tc': len(tables),
                ':pc': page_count
            }
        )

        logger.info(f"Successfully processed {document_id}: {len(high_confidence_kvs)} key-value pairs, {len(tables)} tables")
        return {'statusCode': 200, 'body': f'Processed {document_id}'}

    except Exception as e:
        logger.error(f"Error processing {document_id}: {str(e)}")
        table.update_item(
            Key={'document_id': document_id},
            UpdateExpression='SET processing_status = :status, error_message = :err, processed_at = :ts',
            ExpressionAttributeValues={
                ':status': 'FAILED',
                ':err': str(e),
                ':ts': datetime.now(timezone.utc).isoformat()
            }
        )
        raise


def extract_raw_text(blocks):
    lines = [b['Text'] for b in blocks if b['BlockType'] == 'LINE']
    return '\n'.join(lines)


def extract_key_values(blocks):
    block_map = {b['Id']: b for b in blocks}
    key_values = {}

    for block in blocks:
        if block['BlockType'] == 'KEY_VALUE_SET' and 'KEY' in block.get('EntityTypes', []):
            key_text = get_text_from_block(block, block_map)
            value_block = get_value_block(block, block_map)
            if value_block:
                value_text = get_text_from_block(value_block, block_map)
                confidence = block.get('Confidence', 0)
                if key_text.strip():
                    key_values[key_text.strip()] = {
                        'value': value_text.strip(),
                        'confidence': round(confidence, 2)
                    }

    return key_values


def get_value_block(key_block, block_map):
    for rel in key_block.get('Relationships', []):
        if rel['Type'] == 'VALUE':
            for value_id in rel['Ids']:
                return block_map.get(value_id)
    return None


def get_text_from_block(block, block_map):
    text = ''
    for rel in block.get('Relationships', []):
        if rel['Type'] == 'CHILD':
            for child_id in rel['Ids']:
                child = block_map.get(child_id)
                if child and child['BlockType'] == 'WORD':
                    text += child.get('Text', '') + ' '
    return text


def extract_tables(blocks):
    block_map = {b['Id']: b for b in blocks}
    tables = []

    for block in blocks:
        if block['BlockType'] == 'TABLE':
            table_data = {}
            for rel in block.get('Relationships', []):
                if rel['Type'] == 'CHILD':
                    for cell_id in rel['Ids']:
                        cell = block_map.get(cell_id)
                        if cell and cell['BlockType'] == 'CELL':
                            row_idx = cell.get('RowIndex', 0)
                            col_idx = cell.get('ColumnIndex', 0)
                            cell_text = get_text_from_block(cell, block_map).strip()
                            if row_idx not in table_data:
                                table_data[row_idx] = {}
                            table_data[row_idx][col_idx] = cell_text

            # Convert to list of rows
            if table_data:
                max_row = max(table_data.keys())
                max_col = max(col for row in table_data.values() for col in row.keys())
                structured_table = []
                for r in range(1, max_row + 1):
                    row = []
                    for c in range(1, max_col + 1):
                        row.append(table_data.get(r, {}).get(c, ''))
                    structured_table.append(row)
                tables.append(structured_table)

    return tables

Your pipeline now produces structured, trustworthy extraction results. Tables come back as readable nested lists, low-confidence fields are flagged separately, and every document has a clear audit trail from upload to completion. Next up, you'll extend this pipeline with natural-language queries in the Secret Mission.

Secret mission

Add Natural Language Queries

Your pipeline already extracts tables and forms from documents. But what if you could ask specific questions like 'What is the invoice number?' and get a direct answer? In this extension, you'll add Textract's Queries feature to move from generic extraction to intelligent document understanding.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. All resources in this project fall within the AWS Free Tier, so there are no ongoing costs as long as you stay within the free tier limits.

Resources you used:

  • S3 bucket (free tier: 5 GB storage for 12 months).
  • Lambda function DocEngineProcessor (always free: 1M requests/month).
  • DynamoDB table DocumentMetadata (always free: 25 GB storage, 25 RCU/WCU).
  • IAM role DocEngineRole with inline policy (no cost).
  • Textract usage (free tier: 100 pages/month for AnalyzeDocument for first 3 months).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with document processing.

  • All resources stay within the AWS Free Tier for normal learning usage.
  • Lambda only runs when triggered by an S3 upload. No idle cost.
  • DynamoDB is provisioned at 5 RCU/5 WCU, well within the always-free 25 RCU/25 WCU allocation.
  • Textract free tier lasts 3 months from account creation (100 pages/month for AnalyzeDocument with Forms and Tables).

Pause - I'll come back to this later

Shut down active triggers to prevent accidental usage while keeping your resources intact.

  • Lambda only runs when triggered, so there is no idle compute cost.
  • DynamoDB is provisioned at minimal capacity and stays within free tier.
  • S3 storage costs are negligible for the small number of test documents you uploaded.
  • No action is required to pause. Your pipeline only activates when a new document is uploaded.

Delete - I don't want to use this again

Remove all project resources completely. This is irreversible.

  • Empty and delete the S3 bucket by running:
  • Delete the Lambda function by running:
  • Delete the DynamoDB table by running:
  • Delete the inline policy from the IAM role, then delete the role. First, find the inline policy name:
  • Delete the inline policy (replace the policy name with the one returned above):
  • Delete the IAM role:

That's a Wrap!

That's a Wrap!

Nice work! You've built a fully automated, serverless document extraction pipeline that processes documents the moment they land in your S3 bucket.

You've learned how to:

  • Build an event-driven pipeline connecting S3 event notifications to a Lambda function that automatically triggers on document uploads.
  • Extract structured data (tables, forms, and raw text) from documents using Amazon Textract's AnalyzeDocument API with TABLES and FORMS feature types.
  • Store document metadata and extraction results in DynamoDB with status tracking, confidence scoring, and error handling for production-quality extraction.
  • Secret Mission: Used Textract Queries to ask natural-language questions about documents and get targeted answers stored in DynamoDB.

Ready to quiz yourself?