diff --git a/sfn-entity-resolution-dynamodb-cdk/README.md b/sfn-entity-resolution-dynamodb-cdk/README.md new file mode 100644 index 000000000..56a0f5c75 --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/README.md @@ -0,0 +1,107 @@ +# AWS Step Functions to AWS Entity Resolution to Amazon DynamoDB + +This pattern deploys an automated entity matching pipeline that uses AWS Step Functions to orchestrate AWS Entity Resolution matching jobs. When customer records are uploaded to Amazon S3, Amazon EventBridge triggers the state machine which starts a matching job, polls for completion, and stores match metadata in Amazon DynamoDB. + +Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/sfn-entity-resolution-dynamodb-cdk + +Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example. + +## Requirements + +* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources. +* [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured +* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git) +* [Node.js 18+](https://nodejs.org/en/download/) installed +* [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) installed and bootstrapped + +## Deployment Instructions + +1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository: + ```bash + git clone https://github.com/aws-samples/serverless-patterns + ``` +2. Change directory to the pattern directory: + ```bash + cd serverless-patterns/sfn-entity-resolution-dynamodb-cdk/cdk + ``` +3. Install dependencies: + ```bash + npm install + ``` +4. Deploy the stack: + ```bash + npx cdk deploy + ``` + +## How it works + +This pattern creates an automated entity resolution pipeline: + +1. **Amazon S3** receives customer record files (JSON format) uploaded to the `records/` prefix. +2. **Amazon EventBridge** detects the new object creation and triggers the AWS Step Functions state machine. +3. **AWS Step Functions** orchestrates the matching workflow: + - Calls AWS Entity Resolution `StartMatchingJob` API via SDK integration (no AWS Lambda needed) + - Waits 30 seconds between status polls + - Calls `GetMatchingJob` to check if the job has completed + - On success, stores match job metadata in Amazon DynamoDB + - On failure, the state machine reports the error +4. **AWS Entity Resolution** uses ML-based matching to identify duplicate or related customer records across the dataset. +5. **Amazon DynamoDB** stores the match job results including job ID, status, output path, and completion timestamp. + +## Architecture + +``` +Amazon S3 (upload) --> Amazon EventBridge (Object Created) --> AWS Step Functions (orchestrate) + | + +--> AWS Entity Resolution (StartMatchingJob) + | + +--> Poll (GetMatchingJob) until SUCCEEDED + | + +--> Amazon DynamoDB (store results) +``` + +## Testing + +1. Upload a sample customer records file to the source bucket: + ```bash + # Create sample data + cat > /tmp/customers.json << 'EOF' + {"record_id": "1", "full_name": "John Smith", "email": "john.smith@email.com", "phone": "+1-555-0101", "address": "123 Main St, Seattle, WA 98101"} + {"record_id": "2", "full_name": "J. Smith", "email": "jsmith@email.com", "phone": "555-0101", "address": "123 Main Street, Seattle WA"} + {"record_id": "3", "full_name": "Jane Doe", "email": "jane.doe@company.com", "phone": "+1-555-0202", "address": "456 Oak Ave, Portland, OR 97201"} + EOF + + # Upload to source bucket + aws s3 cp /tmp/customers.json s3://entity-resolution-source--/records/customers.json + ``` + +2. Monitor the AWS Step Functions execution in the AWS Console or via CLI: + ```bash + aws stepfunctions list-executions \ + --state-machine-arn \ + --status-filter RUNNING + ``` + +3. Once complete, check the Amazon DynamoDB table for match results: + ```bash + aws dynamodb scan --table-name EntityMatchResults + ``` + +4. Check matched output in the output bucket: + ```bash + aws s3 ls s3://entity-resolution-output--/matched-results/ --recursive + ``` + +## Cleanup + +> **Warning**: Destroying this stack will delete all data in the Amazon S3 buckets and the Amazon DynamoDB table. Back up any data you need before proceeding. + +```bash +npx cdk destroy +``` + +## Resources + +- [AWS Entity Resolution documentation](https://docs.aws.amazon.com/entityresolution/latest/userguide/what-is-service.html) +- [AWS Step Functions SDK integrations](https://docs.aws.amazon.com/step-functions/latest/dg/supported-services-awssdk.html) +- [Amazon EventBridge Amazon S3 event notifications](https://docs.aws.amazon.com/AmazonS3/latest/userguide/EventBridge.html) diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/.gitignore b/sfn-entity-resolution-dynamodb-cdk/cdk/.gitignore new file mode 100644 index 000000000..c6e6f7a14 --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/.gitignore @@ -0,0 +1,7 @@ +node_modules +cdk.out +cdk.context.json +build +*.js +*.d.ts +!jest.config.js diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/bin/app.ts b/sfn-entity-resolution-dynamodb-cdk/cdk/bin/app.ts new file mode 100644 index 000000000..87a45772b --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/bin/app.ts @@ -0,0 +1,15 @@ +#!/usr/bin/env node +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 +// 2026 + +import * as cdk from 'aws-cdk-lib'; +import { SfnEntityResolutionDynamodbStack } from '../lib/sfn-entity-resolution-dynamodb-stack'; + +const app = new cdk.App(); +new SfnEntityResolutionDynamodbStack(app, 'SfnEntityResolutionDynamodbStack', { + env: { + account: process.env.CDK_DEFAULT_ACCOUNT, + region: process.env.CDK_DEFAULT_REGION, + }, +}); diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/cdk.json b/sfn-entity-resolution-dynamodb-cdk/cdk/cdk.json new file mode 100644 index 000000000..56ed0cad7 --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/cdk.json @@ -0,0 +1,6 @@ +{ + "app": "npx ts-node --prefer-ts-exts bin/app.ts", + "context": { + "@aws-cdk/customresources:installLatestAwsSdkDefault": false + } +} diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/lib/sfn-entity-resolution-dynamodb-stack.ts b/sfn-entity-resolution-dynamodb-cdk/cdk/lib/sfn-entity-resolution-dynamodb-stack.ts new file mode 100644 index 000000000..7ce1b004b --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/lib/sfn-entity-resolution-dynamodb-stack.ts @@ -0,0 +1,280 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 +// 2026 + +import * as cdk from 'aws-cdk-lib'; +import { Construct } from 'constructs'; +import * as s3 from 'aws-cdk-lib/aws-s3'; +import * as dynamodb from 'aws-cdk-lib/aws-dynamodb'; +import * as events from 'aws-cdk-lib/aws-events'; +import * as events_targets from 'aws-cdk-lib/aws-events-targets'; +import * as sfn from 'aws-cdk-lib/aws-stepfunctions'; +import * as tasks from 'aws-cdk-lib/aws-stepfunctions-tasks'; +import * as iam from 'aws-cdk-lib/aws-iam'; +import * as glue from 'aws-cdk-lib/aws-glue'; + +export class SfnEntityResolutionDynamodbStack extends cdk.Stack { + constructor(scope: Construct, id: string, props?: cdk.StackProps) { + super(scope, id, props); + + // Source data bucket (upload CSV/JSON records here for matching) + const sourceBucket = new s3.Bucket(this, 'SourceDataBucket', { + bucketName: `entity-resolution-source-${cdk.Aws.ACCOUNT_ID}-${cdk.Aws.REGION}`, + removalPolicy: cdk.RemovalPolicy.DESTROY, + autoDeleteObjects: true, + encryption: s3.BucketEncryption.S3_MANAGED, + eventBridgeEnabled: true, + }); + + // Output bucket (Entity Resolution writes matched results here) + const outputBucket = new s3.Bucket(this, 'OutputBucket', { + bucketName: `entity-resolution-output-${cdk.Aws.ACCOUNT_ID}-${cdk.Aws.REGION}`, + removalPolicy: cdk.RemovalPolicy.DESTROY, + autoDeleteObjects: true, + encryption: s3.BucketEncryption.S3_MANAGED, + }); + + // DynamoDB table for matched entity results + const matchResultsTable = new dynamodb.Table(this, 'MatchResultsTable', { + tableName: 'EntityMatchResults', + partitionKey: { name: 'matchId', type: dynamodb.AttributeType.STRING }, + sortKey: { name: 'sourceRecordId', type: dynamodb.AttributeType.STRING }, + billingMode: dynamodb.BillingMode.PAY_PER_REQUEST, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }); + + // AWS Glue database and table for Entity Resolution schema mapping + const glueDatabase = new glue.CfnDatabase(this, 'GlueDatabase', { + catalogId: cdk.Aws.ACCOUNT_ID, + databaseInput: { + name: 'entity_resolution_db', + description: 'Database for AWS Entity Resolution schema mapping', + }, + }); + + const glueTable = new glue.CfnTable(this, 'GlueTable', { + catalogId: cdk.Aws.ACCOUNT_ID, + databaseName: 'entity_resolution_db', + tableInput: { + name: 'customer_records', + description: 'Customer records for entity matching', + storageDescriptor: { + columns: [ + { name: 'record_id', type: 'string' }, + { name: 'full_name', type: 'string' }, + { name: 'email', type: 'string' }, + { name: 'phone', type: 'string' }, + { name: 'address', type: 'string' }, + ], + location: `s3://${sourceBucket.bucketName}/records/`, + inputFormat: 'org.apache.hadoop.mapred.TextInputFormat', + outputFormat: 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat', + serdeInfo: { + serializationLibrary: 'org.openx.data.jsonserde.JsonSerDe', + }, + }, + tableType: 'EXTERNAL_TABLE', + }, + }); + glueTable.addDependency(glueDatabase); + + // IAM Role for Entity Resolution + const entityResolutionRole = new iam.Role(this, 'EntityResolutionRole', { + assumedBy: new iam.ServicePrincipal('entityresolution.amazonaws.com'), + description: 'Role for AWS Entity Resolution to access source and output data', + }); + + sourceBucket.grantRead(entityResolutionRole); + outputBucket.grantWrite(entityResolutionRole); + + entityResolutionRole.addToPolicy(new iam.PolicyStatement({ + actions: [ + 'glue:GetTable', + 'glue:GetTableVersion', + 'glue:GetTableVersions', + 'glue:GetPartitions', + 'glue:GetDatabase', + 'glue:BatchGetPartition', + ], + resources: [ + `arn:aws:glue:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:catalog`, + `arn:aws:glue:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:database/entity_resolution_db`, + `arn:aws:glue:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:table/entity_resolution_db/*`, + ], + })); + + // Entity Resolution Schema Mapping + const schemaMapping = new cdk.CfnResource(this, 'SchemaMapping', { + type: 'AWS::EntityResolution::SchemaMapping', + properties: { + SchemaName: 'CustomerSchemaMapping', + Description: 'Schema mapping for customer record matching', + MappedInputFields: [ + { FieldName: 'record_id', Type: 'UNIQUE_ID' }, + { FieldName: 'full_name', Type: 'NAME', SubType: 'FULL' }, + { FieldName: 'email', Type: 'EMAIL_ADDRESS' }, + { FieldName: 'phone', Type: 'PHONE_NUMBER' }, + { FieldName: 'address', Type: 'ADDRESS', SubType: 'FULL' }, + ], + }, + }); + + // Entity Resolution Matching Workflow + const matchingWorkflow = new cdk.CfnResource(this, 'MatchingWorkflow', { + type: 'AWS::EntityResolution::MatchingWorkflow', + properties: { + WorkflowName: 'CustomerMatchingWorkflow', + Description: 'Match customer records to identify duplicate entities', + InputSourceConfig: [ + { + InputSourceARN: `arn:aws:glue:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:table/entity_resolution_db/customer_records`, + SchemaArn: schemaMapping.getAtt('SchemaArn'), + }, + ], + OutputSourceConfig: [ + { + OutputS3Path: `s3://${outputBucket.bucketName}/matched-results/`, + Output: [ + { Name: 'record_id', Hashed: false }, + { Name: 'full_name', Hashed: false }, + { Name: 'email', Hashed: false }, + ], + }, + ], + ResolutionTechniques: { + ResolutionType: 'ML_MATCHING', + }, + RoleArn: entityResolutionRole.roleArn, + }, + }); + matchingWorkflow.addDependency(glueTable); + // Ensure IAM role policy is fully propagated before Entity Resolution validates access + const roleDefaultPolicy = entityResolutionRole.node.findChild('DefaultPolicy') as iam.Policy; + matchingWorkflow.node.addDependency(roleDefaultPolicy); + + // Step Functions state machine + // Step 1: Start the matching job + const startMatchingJob = new tasks.CallAwsService(this, 'StartMatchingJob', { + service: 'entityresolution', + action: 'startMatchingJob', + parameters: { + WorkflowName: 'CustomerMatchingWorkflow', + }, + iamResources: [`arn:aws:entityresolution:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:matchingworkflow/CustomerMatchingWorkflow`], + resultPath: '$.matchingJob', + }); + + // Step 2: Wait for job completion + const waitForJob = new sfn.Wait(this, 'WaitForJobCompletion', { + time: sfn.WaitTime.duration(cdk.Duration.seconds(30)), + }); + + // Step 3: Check job status + const getJobStatus = new tasks.CallAwsService(this, 'GetMatchingJob', { + service: 'entityresolution', + action: 'getMatchingJob', + parameters: { + WorkflowName: 'CustomerMatchingWorkflow', + 'JobId.$': '$.matchingJob.JobId', + }, + iamResources: [`arn:aws:entityresolution:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT_ID}:matchingworkflow/CustomerMatchingWorkflow`], + resultPath: '$.jobStatus', + }); + + // Step 4: Store results in DynamoDB + const storeResults = new tasks.DynamoPutItem(this, 'StoreMatchResults', { + table: matchResultsTable, + item: { + matchId: tasks.DynamoAttributeValue.fromString(sfn.JsonPath.stringAt('$.matchingJob.JobId')), + sourceRecordId: tasks.DynamoAttributeValue.fromString(sfn.JsonPath.format('job-{}', sfn.JsonPath.stringAt('$.matchingJob.JobId'))), + status: tasks.DynamoAttributeValue.fromString(sfn.JsonPath.stringAt('$.jobStatus.Status')), + outputPath: tasks.DynamoAttributeValue.fromString( + sfn.JsonPath.format('s3://{}/matched-results/', outputBucket.bucketName) + ), + completedAt: tasks.DynamoAttributeValue.fromString(sfn.JsonPath.stringAt('$$.State.EnteredTime')), + }, + resultPath: '$.dynamoResult', + }); + + // Job success path + const jobSucceeded = new sfn.Succeed(this, 'MatchingJobComplete', { + comment: 'Entity matching job completed successfully', + }); + + // Job failed path + const jobFailed = new sfn.Fail(this, 'MatchingJobFailed', { + error: 'MatchingJobFailed', + cause: 'The AWS Entity Resolution matching job failed', + }); + + // Check status choice + const isJobComplete = new sfn.Choice(this, 'IsJobComplete') + .when( + sfn.Condition.stringEquals('$.jobStatus.Status', 'SUCCEEDED'), + storeResults.next(jobSucceeded) + ) + .when( + sfn.Condition.stringEquals('$.jobStatus.Status', 'FAILED'), + jobFailed + ) + .otherwise(waitForJob); + + // Wire the state machine + const definition = startMatchingJob + .next(waitForJob) + .next(getJobStatus) + .next(isJobComplete); + + const stateMachine = new sfn.StateMachine(this, 'EntityResolutionStateMachine', { + stateMachineName: 'EntityResolutionOrchestrator', + definitionBody: sfn.DefinitionBody.fromChainable(definition), + timeout: cdk.Duration.hours(2), + }); + + // EventBridge rule: trigger when new data uploaded to source bucket + const uploadRule = new events.Rule(this, 'NewDataUploadRule', { + ruleName: 'EntityResolutionNewDataTrigger', + description: 'Triggers entity matching when new records are uploaded to the source Amazon S3 bucket', + eventPattern: { + source: ['aws.s3'], + detailType: ['Object Created'], + detail: { + bucket: { + name: [sourceBucket.bucketName], + }, + object: { + key: [{ prefix: 'records/' }], + }, + }, + }, + }); + + uploadRule.addTarget(new events_targets.SfnStateMachine(stateMachine)); + + // Outputs + new cdk.CfnOutput(this, 'SourceBucketName', { + value: sourceBucket.bucketName, + description: 'Upload customer records (JSON) to the records/ prefix in this bucket', + }); + + new cdk.CfnOutput(this, 'OutputBucketName', { + value: outputBucket.bucketName, + description: 'Matched entity results are written here by AWS Entity Resolution', + }); + + new cdk.CfnOutput(this, 'MatchResultsTableName', { + value: matchResultsTable.tableName, + description: 'Amazon DynamoDB table storing match job metadata and results', + }); + + new cdk.CfnOutput(this, 'StateMachineArn', { + value: stateMachine.stateMachineArn, + description: 'AWS Step Functions state machine orchestrating entity matching', + }); + + new cdk.CfnOutput(this, 'MatchingWorkflowName', { + value: 'CustomerMatchingWorkflow', + description: 'AWS Entity Resolution matching workflow name', + }); + } +} diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/package.json b/sfn-entity-resolution-dynamodb-cdk/cdk/package.json new file mode 100644 index 000000000..9207178e8 --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/package.json @@ -0,0 +1,24 @@ +{ + "name": "sfn-entity-resolution-dynamodb-cdk", + "version": "1.0.0", + "description": "AWS Step Functions orchestrates AWS Entity Resolution matching jobs triggered by Amazon S3 uploads, storing results in Amazon DynamoDB", + "bin": { + "app": "bin/app.ts" + }, + "scripts": { + "build": "tsc", + "synth": "cdk synth", + "deploy": "cdk deploy", + "destroy": "cdk destroy" + }, + "dependencies": { + "aws-cdk-lib": "^2.185.0", + "constructs": "^10.0.0" + }, + "devDependencies": { + "typescript": "~5.4.0", + "@types/node": "^20.0.0", + "ts-node": "^10.9.0", + "aws-cdk": "^2.185.0" + } +} diff --git a/sfn-entity-resolution-dynamodb-cdk/cdk/tsconfig.json b/sfn-entity-resolution-dynamodb-cdk/cdk/tsconfig.json new file mode 100644 index 000000000..bd6d3bc33 --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/cdk/tsconfig.json @@ -0,0 +1,25 @@ +{ + "compilerOptions": { + "target": "ES2020", + "module": "commonjs", + "lib": ["ES2020"], + "declaration": true, + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "noImplicitThis": true, + "alwaysStrict": true, + "noUnusedLocals": false, + "noUnusedParameters": false, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": false, + "inlineSourceMap": true, + "inlineSources": true, + "experimentalDecorators": true, + "strictPropertyInitialization": false, + "typeRoots": ["./node_modules/@types"], + "outDir": "./build", + "rootDir": "." + }, + "exclude": ["node_modules", "build"] +} diff --git a/sfn-entity-resolution-dynamodb-cdk/example-pattern.json b/sfn-entity-resolution-dynamodb-cdk/example-pattern.json new file mode 100644 index 000000000..134b7ca5f --- /dev/null +++ b/sfn-entity-resolution-dynamodb-cdk/example-pattern.json @@ -0,0 +1,108 @@ +{ + "title": "AWS Step Functions to AWS Entity Resolution to Amazon DynamoDB", + "description": "Orchestrate AWS Entity Resolution matching jobs with AWS Step Functions, triggered by Amazon S3 uploads, storing results in Amazon DynamoDB", + "language": "TypeScript", + "level": "300", + "framework": "AWS CDK", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern deploys an automated entity matching pipeline using AWS Step Functions.", + "Customer records uploaded to Amazon S3 trigger an Amazon EventBridge rule.", + "AWS Step Functions orchestrates the AWS Entity Resolution matching job using native SDK integration.", + "The state machine polls for job completion and stores match metadata in Amazon DynamoDB.", + "No AWS Lambda functions are needed for orchestration thanks to direct SDK service integrations." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/sfn-entity-resolution-dynamodb-cdk", + "templateURL": "serverless-patterns/sfn-entity-resolution-dynamodb-cdk", + "projectFolder": "sfn-entity-resolution-dynamodb-cdk", + "templateFile": "cdk/lib/sfn-entity-resolution-dynamodb-stack.ts" + } + }, + "resources": { + "bullets": [ + { + "text": "AWS Entity Resolution - ML-powered entity matching", + "link": "https://docs.aws.amazon.com/entityresolution/latest/userguide/what-is-service.html" + }, + { + "text": "AWS Step Functions - Serverless workflow orchestration", + "link": "https://docs.aws.amazon.com/step-functions/latest/dg/welcome.html" + }, + { + "text": "Amazon EventBridge - Serverless event bus", + "link": "https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-what-is.html" + } + ] + }, + "deploy": { + "text": [ + "cd sfn-entity-resolution-dynamodb-cdk/cdk", + "npm install", + "npx cdk deploy" + ] + }, + "cleanup": { + "text": [ + "npx cdk destroy" + ] + }, + "authors": [ + { + "name": "Nithin Chandran R", + "bio": "Technical Account Manager at AWS", + "linkedin": "nithin-chandran-r" + } + ], + "patternArch": { + "icon1": { + "x": 15, + "y": 50, + "service": "s3", + "label": "Amazon S3" + }, + "icon2": { + "x": 35, + "y": 50, + "service": "eventbridge", + "label": "Amazon EventBridge" + }, + "icon3": { + "x": 55, + "y": 50, + "service": "sfn", + "label": "AWS Step Functions" + }, + "icon4": { + "x": 75, + "y": 50, + "service": "entity-resolution", + "label": "AWS Entity Resolution" + }, + "icon5": { + "x": 90, + "y": 50, + "service": "dynamodb", + "label": "Amazon DynamoDB" + }, + "line1": { + "from": "icon1", + "to": "icon2" + }, + "line2": { + "from": "icon2", + "to": "icon3" + }, + "line3": { + "from": "icon3", + "to": "icon4" + }, + "line4": { + "from": "icon3", + "to": "icon5" + } + } +}