So laden Sie Streaming-Daten aus Amazon DynamoDB

Sie können AWS Lambda es verwenden, um Daten von Amazon DynamoDB an Ihre OpenSearch Service-Domain zu senden. Neue in der Datenbanktabelle eintreffende Daten lösen eine Ereignisbenachrichtigung an Lambda aus, wodurch Ihr benutzerdefinierte Code zum Durchführen der Indizierung ausgeführt wird.

Voraussetzungen

Zum Fortfahren benötigen Sie die folgenden Ressourcen.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "es:ESHttpPost",
        "es:ESHttpPut",
        "dynamodb:DescribeStream",
        "dynamodb:GetRecords",
        "dynamodb:GetShardIterator",
        "dynamodb:ListStreams",
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "*"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

So erstellen Sie die Lambda-Funktion:

Befolgen Sie die Anweisungen in Erstellen des Lambda-Bereitstellungspakets. Erstellen Sie jedoch ein Verzeichnis mit dem Namen ddb-to-opensearch und verwenden Sie den folgenden Code für sample.py:

import boto3
import requests
from requests_aws4auth import AWS4Auth

region = '' # e.g. us-east-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype + '/'

headers = { "Content-Type": "application/json" }

def handler(event, context):
    count = 0
    for record in event['Records']:
        # Get the primary key for use as the OpenSearch ID
        id = record['dynamodb']['Keys']['id']['S']

        if record['eventName'] == 'REMOVE':
            r = requests.delete(url + id, auth=awsauth)
        else:
            document = record['dynamodb']['NewImage']
            r = requests.put(url + id, auth=awsauth, json=document, headers=headers)
        count += 1
    return str(count) + ' records processed.'

Bearbeiten Sie die Variablen für region und host.

Installieren Sie pip, falls noch nicht geschehen, und verwenden Sie dann die folgenden Befehle, um Ihre Abhängigkeiten zu installieren:

cd ddb-to-opensearch

pip install --target ./package requests
pip install --target ./package requests_aws4auth

Folgen Sie dann den Anweisungen unterSo erstellen Sie die Lambda-Funktion:, geben Sie jedoch die IAM Rolle von Voraussetzungen und die folgenden Einstellungen für den Auslöser an:

Weitere Informationen finden Sie unter Verarbeiten neuer Elemente mit DynamoDB-Streams und Lambda im Amazon-DynamoDB-Entwicklerhandbuch.

Zu diesem Zeitpunkt verfügen Sie über einen vollständigen Satz von Ressourcen: eine DynamoDB-Tabelle für Ihre Quelldaten, einen DynamoDB-Stream mit Änderungen an der Tabelle, eine Funktion, die nach Änderungen Ihrer Quelldaten ausgeführt wird und diese Änderungen indexiert, und eine OpenSearch Service-Domäne für die Suche und Visualisierung.

Lambda-Funktion testen

Nachdem Sie die Funktion erstellt haben, können Sie sie testen, indem Sie ein neues Element zur DynamoDB-Tabelle mithilfe der AWS CLI hinzufügen:

aws dynamodb put-item --table-name test --item '{"director": {"S": "Kevin Costner"},"id": {"S": "00001"},"title": {"S": "The Postman"}}' --region us-west-1

Überprüfen Sie dann mithilfe der OpenSearch Servicekonsole oder der OpenSearch Dashboards, ob das Dokument ein Dokument enthält. lambda-index Sie können außerdem die folgenden Anforderung verwenden:

GET https://domain-name/lambda-index/_doc/00001
{
    "_index": "lambda-index",
    "_type": "_doc",
    "_id": "00001",
    "_version": 1,
    "found": true,
    "_source": {
        "director": {
            "S": "Kevin Costner"
        },
        "id": {
            "S": "00001"
        },
        "title": {
            "S": "The Postman"
        }
    }
}