從 Amazon DynamoDB 中載入串流資料 - Amazon OpenSearch Service
先決條件建立 Lambda 函式測試 Lambda 函數

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

從 Amazon DynamoDB 中載入串流資料

您可以使用 從 Amazon DynamoDB AWS Lambda 將資料傳送至 OpenSearch Service 網域。送達資料庫資料表的新資料會觸發 Lambda 的事件通知，然後執行您的自訂程式碼以執行索引。

先決條件

繼續之前，您必須準備好以下資源。

先決條件 描述
DynamoDB 資料表

表格中包含您的來源資料。如需詳細資訊，請參閱 Amazon DynamoDB 開發人員指南中的 DynamoDB 資料表上的基本操作。

資料表必須位於和 OpenSearch Service 網域相同的區域，並將串流設定為 New image (新映像)。如需進一步了解，請參閱啟用串流。
OpenSearch Service 域 您的 Lambda 函數處理資料後的資料目的地。如需詳細資訊，請參閱 建立 OpenSearch Service 網域。
IAM role (IAM 角色)

此角色必須擁有基本 OpenSearch Service、DynamoDB 和 Lambda 執行許可，例如下列項目：

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

角色必須具有下列信任關係：

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

如需進一步了解，請參閱 IAM 使用者指南中的建立 IAM 角色。

建立 Lambda 函式

遵循建立 Lambda 部署套件中的指示，但要建立名為 ddb-to-opensearch 的目錄，並使用以下適用於 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.'

編輯 region 和 host 的變數。

安裝 pip (如果您尚未安裝的話)，然後使用下列命令安裝相依項目：

cd ddb-to-opensearch

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

然後，遵循建立 Lambda 函式中的指示，但要從先決條件指定 IAM 角色和下列用於觸發的設定：

  • 資料表：您的 DynamoDB 資料表

  • 批次大小：100

  • 開始位置：水平修剪

如需進一步了解，請參閱 Amazon DynamoDB 開發人員指南中的使用 DynamoDB Streams 和 Lambda 來處理新項目。

此時，您有一整組的資源：來源資料的 DynamoDB 資料表、資料表的 DynamoDB 串流變更、在您的來源資料變更並索引這些變更後所執行的函數，以及可搜尋和視覺化的 OpenSearch Service 網域。

測試 Lambda 函數

在建立函數後，您可以進行測試，方法是將新項目新增至使用 AWS CLI的 DynamoDB 資料表：

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

然後使用 OpenSearch Service 主控台或 OpenSearch Dashboards 來確認 lambda-index 包含一個文件。您也可以使用以下請求：

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