Caricamento di dati in streaming da una tabella Amazon DynamoDB

Puoi utilizzarlo AWS Lambda per inviare dati al tuo dominio di OpenSearch servizio da Amazon DynamoDB. I nuovi dati che arrivano nella tabella di database attivano una notifica eventi per Lambda, che quindi esegue il codice personalizzato per eseguire l'indicizzazione.

Prerequisiti

Prima di procedere, devi disporre delle risorse indicate di seguito.

{
  "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"
    }
  ]
}

Creazione della funzione Lambda

Procedi come descritto in Creazione il pacchetto di implementazione Lambda, ma crea una directory denominata ddb-to-opensearch e utilizza il codice seguente per 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.'

Modifica le variabili per region e host.

Installare pip, se non è già stato fatto, quindi utilizzare i seguenti comandi per installare le dipendenze:

cd ddb-to-opensearch

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

Quindi segui le istruzioni riportate inCreazione della funzione Lambda, ma specifica il IAM ruolo di Prerequisiti e le seguenti impostazioni per il trigger:

Per ulteriori informazioni, consultare Elaborazione di nuovi elementi con DynamoDB Streams e Lambdanella Guida per gli sviluppatori di Amazon DynamoDB.

A questo punto, hai a disposizione un set completo di risorse: una tabella DynamoDB per i dati di origine, un flusso DynamoDB di modifiche alla tabella, una funzione che viene eseguita dopo le modifiche dei dati di origine e indicizza tali modifiche e un dominio di servizio per la ricerca e la visualizzazione. OpenSearch

Test della funzione Lambda

Una volta creata la funzione, è possibile eseguirne il test aggiungendo un nuovo elemento alla tabella DynamoDB utilizzando la AWS CLI:

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

Utilizza quindi la console di OpenSearch servizio o le OpenSearch dashboard per verificare che contenga un documento. lambda-index Puoi inoltre utilizzare la seguente richiesta:

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"
        }
    }
}