使用AWS DMS迁移MySQL(Aurora)至OpenSearch：合并表生成单索引可行吗？

我需要使用AWS DMS将公司的MySQL(Aurora)数据库迁移至AWS OpenSearch数据库。根据官方文档，DMS会为每个关系型数据库表创建独立索引，但我希望通过关联表生成唯一索引。例如，假设有一对多关联的Order表和OrderLine表，能否将数据迁移为如下格式的文档？

"hits": [
      {
        "_index": "orders",
        "_id": "12345",
        "_score": 1,
        "_source": {
          "orderId": "12345",
          "orderDate": "2023-10-23T21:05:37.263",
          "store": "Store-1",
          "orderLines": [
            {
              "lineNumber": 1,
              "productName": "Product-1",
              "quantity": 1,
              "price": 119.99
            },
            {
              "lineNumber": 2,
              "productName": "Product-2",
              "quantity": 1,
              "price": 39.99
            }
       ]
    }
}

可以实现，但AWS DMS原生不支持直接关联关系型表生成嵌套结构的OpenSearch文档，需要结合额外的AWS服务来完成。以下是几种可行的方案：

方案1：DMS + Lambda 实时同步处理 #

适合需要实时增量同步的场景：

• 第一步：配置DMS任务，将Aurora中的Order和OrderLine表分别同步到OpenSearch的临时索引（比如orders_raw和orderlines_raw）。
• 第二步：给这两个临时索引配置OpenSearch触发器，当有新数据写入时触发Lambda函数。
• 第三步：在Lambda中编写业务逻辑：
  • 若触发源是orders_raw的订单数据：直接在目标orders索引中创建或更新对应的订单文档。
  • 若触发源是orderlines_raw的订单行数据：通过orderId匹配到orders索引中的对应订单文档，使用OpenSearch的Update API将订单行数据追加到orderLines数组中。
• 注意事项：
  • 处理幂等性：给订单行数据加唯一标识，避免重复写入。
  • 处理时序问题：如果订单行先于订单写入，Lambda中需加重试逻辑，确保订单文档存在后再追加数据。
  • 支持更新/删除：当订单行数据更新或删除时，要在Lambda中找到对应数组元素进行修改或移除。

方案2：DMS + S3 + OpenSearch Ingest Pipeline 批量处理 #

适合全量迁移或批量增量同步的场景：

• 第一步：配置DMS任务，将Aurora中的Order和OrderLine表数据全量导出到S3存储桶，按表生成JSON格式的数据文件（比如orders/和orderlines/前缀下的文件）。
• 第二步：使用OpenSearch的Ingest Pipeline进行数据关联：
  • 先读取S3中的订单数据作为主文档。
  • 通过lookup处理器关联S3中的订单行数据，按orderId匹配后，将订单行数据组装成嵌套数组orderLines。
  • 或者用Lambda读取S3中的两个数据集，在内存中完成关联后，再批量写入OpenSearch的orders索引。
• 注意事项：
  • 若需要增量同步，可以配置S3事件通知，当DMS写入新的增量文件时触发Lambda处理。

方案3：AWS Glue ETL转换 #

适合复杂关联逻辑或需要定期同步的场景：

• 第一步：在Glue中创建爬虫，抓取Aurora中Order和OrderLine表的元数据，生成对应的Glue Data Catalog表。
• 第二步：编写Glue Spark Job，用SQL完成数据关联和嵌套转换：

  SELECT 
    o.orderId,
    o.orderDate,
    o.store,
    collect_list(
      struct(
        l.lineNumber,
        l.productName,
        l.quantity,
        l.price
      )
    ) AS orderLines
  FROM orders o
  JOIN orderlines l ON o.orderId = l.orderId
  GROUP BY o.orderId, o.orderDate, o.store
  

• 第三步：将Glue Job处理后的数据集直接写入OpenSearch的orders索引。
• 注意事项：
  • 配置Glue Job的并行度，提升大数据量下的处理效率。
  • 支持增量同步：通过Glue的书签功能，只同步上次任务之后更新的数据。

• 提前创建OpenSearch索引映射：将orderLines字段设置为nested类型，避免后续嵌套查询出现问题。
• 数据一致性校验：迁移完成后，抽样对比Aurora和OpenSearch中的数据，确保关联后的嵌套数据准确。

内容的提问来源于stack exchange，提问作者NecmiK