將 OpenSearch 擷取管道與 Amazon DynamoDB 使用

您可以將 OpenSearch 擷取管道與 DynamoDB 搭配使用，將 DynamoDB 表格事件 (例如建立、更新和刪除) 串流至 Amazon OpenSearch 服務網域和集合。 OpenSearch 擷取管道整合了變更資料擷取 (CDC) 基礎架構，以提供高規模、低延遲的方式來持續從 DynamoDB 表格串流資料。

您可以透過兩種方式使用 DynamoDB 做為處理資料的來源，無論是否有完整的初始快照。

完整初始快照是 DynamoDB 使用point-in-time 復原 (PITR) 功能所採用的資料表備份。DynamoDB 此快照上傳到 Amazon S3。從那裡， OpenSearch 擷取管線會將其傳送到網域中的一個索引，或將其分割為網域中的多個索引。為了保持 DynamoDB 中的資料 OpenSearch 一致性，管線會將 DynamoDB 表中的所有建立、更新和刪除事件與儲存在索引中的文件同步。 OpenSearch

使用完整初始快照時，擷取管道會先 OpenSearch 擷取快照，然後開始從 DynamoDB Streams 讀取資料。它最終可以追趕並維持 DynamoDB 和. 之間近乎即時的資料一致性。 OpenSearch選擇此選項時，您必須在表格上同時啟用PITR和 DynamoDB 串流。

您也可以使用與 DynamoDB 的 OpenSearch 擷取整合，在沒有快照的情況下串流事件。如果您已經擁有其他機制的完整快照，或者您只想從 DynamoDB 資料表中 DynamoDB Streams 目前的事件，請選擇此選項。選擇此選項時，您只需要在表格上啟用 DynamoDB 串流即可。

如需有關此整合的詳細資訊，請參閱Amazon DynamoDB 開發人員指南中的 DynamoDB 與 Amazon OpenSearch 服務零ETL整合。

必要條件

若要設定管線，您必須啟用 DynamoDB 串流的 DynamoDB 表格。您的串流應該使用NEW_IMAGE串流檢視類型。不過，NEW_AND_OLD_IMAGES如果此串流檢視類型符合您的使用案例， OpenSearch 擷取管線也可以串流事件。

如果您使用的是快照，您也必須在資料表上啟用 point-in-time 復原功能。如需詳細資訊，請參閱 Amazon DynamoDB 開發人員指南中的建立表格、啟用 point-in-time 復原和啟用串流。

步驟 1：設定管線角色

設定 DynamoDB 表之後，請設定要在管線組態中使用的管線角色，並在角色中新增下列 DynamoDB 權限：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "allowRunExportJob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeTable",
                "dynamodb:DescribeContinuousBackups",
                "dynamodb:ExportTableToPointInTime"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table"
            ]
        },
        {
            "Sid": "allowCheckExportjob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeExport"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/export/*"
            ]
        },
        {
            "Sid": "allowReadFromStream",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeStream",
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/stream/*"
            ]
        },
        {
            "Sid": "allowReadAndWriteToS3ForExport",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": [
                "arn:aws:s3:::my-bucket/{exportPath}/*"
            ]
        }
    ]
}

您也可以使用 AWS KMS 客戶管理的金鑰來加密匯出資料檔案。若要解密匯出的物件，s3_sse_kms_key_id請以下列格式在管線的匯出組態中指定金鑰 ID：arn:aws:kms:us-west-2:{account-id}:key/my-key-id下列原則包含使用客戶管理金鑰的必要權限：

{
    "Sid": "allowUseOfCustomManagedKey",
    "Effect": "Allow",
    "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt"
    ],
    "Resource": arn:aws:kms:us-west-2:{account-id}:key/my-key-id
}

步驟 2：建立管道

然後，您可以如下所示設定 OpenSearch 擷取管道，該管道會將 DynamoDB 指定為來源。此範例管道會從table-aPITR快照擷取資料，接著是 DynamoDB Streams 中的事件。的開始位置LATEST表示管道應該從 DynamoDB Streams 讀取最新資料。

version: "2"
cdc-pipeline:
  source:
    dynamodb:
      tables:
      - table_arn: "arn:aws:dynamodb:us-west-2:{account-id}:table/table-a"  
        export:
          s3_bucket: "my-bucket"
          s3_prefix: "export/"
        stream:
          start_position: "LATEST"
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::{account-id}:role/pipeline-role"
  sink:
  - opensearch:
      hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"]
      index: "${getMetadata(\"table_name\")}"
      index_type: custom
      normalize_index: true
      document_id: "${getMetadata(\"primary_key\")}"
      action: "${getMetadata(\"opensearch_action\")}"
      document_version: "${getMetadata(\"document_version\")}"
      document_version_type: "external"

您可以使用預先設定的 DynamoDB 藍圖來建立此管道。如需詳細資訊，請參閱使用藍圖建立管道。

資料一致性

OpenSearch 擷取支援確 end-to-end 認，以確保資料耐久性。管道讀取快照或串流時，會動態建立分割區以進行 parallel 處理。管道會在擷取 OpenSearch 網域或集合中的所有記錄後收到確認後，將分割區標示為完整。

如果您想要內嵌至 OpenSearch 無伺服器搜尋集合，可以在管道中產生文件 ID。如果您想要導入 OpenSearch 無伺服器時間序列集合，請注意，管線不會產生文件 ID。

OpenSearch 擷取管線也會將傳入的事件動作對應至對應的大量索引動作，以協助擷取文件。這樣可以保持資料一致，以便 DynamoDB 中的每個資料變更都會與中的對應文件變更協調。 OpenSearch

對映資料類型

OpenSearch 服務會動態地將每個傳入文件中的資料類型對應至 DynamoDB 中的對應資料類型。下表顯示了 OpenSearch 服務如何自動映射各種數據類型。

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberAttribute": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "ImageData": {
     "type": "binary"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "AdditionalDescriptions": {
     "properties": {
      "PartType": {
       "type": "keyword"
      }
     }
    }
   }
  }
 }
}

建議您在 OpenSearch 擷取管線中設定無效字母佇列 (DLQ)。如果您已設定佇列， OpenSearch Service 會將由於動態對應失敗而無法擷取的所有失敗文件傳送至佇列。

如果自動對映失敗，您可以在管線組態template_content中使用template_type和來定義明確的對應規則。或者，您可以在啟動管道之前，直接在搜尋網域或集合中建立對應範本。

限制

為 DynamoDB 設定 OpenSearch 擷取管線時，請考慮下列限制：

建議 CloudWatch DynamoDB 示

建議您使用下列 CloudWatch 指標來監控擷取管線的效能。這些量度可協助您識別從匯出中處理的資料量、從串流處理的事件數量、處理匯出和串流事件時的錯誤，以及寫入目標的文件數量。您可以設定 CloudWatch 警示，以便在其中一個量度超過指定值超過指定的時間長度時執行動作。