将 JSON 数据从 Oracle 的 CLOB 迁移到 Amazon Aurora Postgre

从 Oracle CLOB 迁移 JSON 数据到 Amazon Aurora PostgreSQL 和 Amazon RDS for PostgreSQL

发布日期 2024年1月17日 作者 Sumana Yanamandra Ramu Jagini Shivam Rastogi

关键要点

迁移数据库从 Oracle 到 Amazon Aurora PostgreSQL 兼容版 或 Amazon RDS for PostgreSQL 时，处理 JSON 数据会面临独特的挑战。Oracle 的 CLOB 数据类型可以包含格式正确的 JSON 和无效的 JSON，而 PostgreSQL 中的 JSON 和 JSONB 数据类型要求 JSON 数据必须符合 RFC 8259 的格式。因此，将存储在 Oracle CLOB 数据类型中的数据迁移到 Amazon Aurora PostgreSQL 或 Amazon RDS for PostgreSQL 的 JSON 和 JSONB 数据类型时，需要仔细检查数据质量和完整性。这段迁移过程同时要处理格式正确和不正确的 JSON 数据。无效的 JSON 可能包含语法和语义错误，例如缺失的括号、遗漏的逗号或结构不正确的格式。如果试图将其插入 JSON 和 JSONB 列，PostgreSQL 将抛出错误。成功迁移这些表意味着要解决和纠正这些数据不一致性。

在本文中，我们展示了使用 AWS 数据库迁移服务 (AWS DMS) 和 PL/pgSQL 程序将 JSON 数据从 Oracle 迁移到 Amazon Aurora PostgreSQL 兼容版 或 Amazon RDS for PostgreSQL 的过程。根据本文提到的方法，可以识别并隔离 CLOB 列中存储的无效 JSON 数据及其相应的主键值。

虽然可以使用 AWS DMS 来简化迁移过程，但在迁移存储在源 Oracle 的 CLOB 列中的无效 JSON 记录时，需要特别谨慎，因为 PostgreSQL 的 JSON 或 JSONB 数据类型会拒绝无效的 JSON。因此，该文章中提到的 PL/pgSQL 程序将是一个补充解决方案。AWS DMS 简化了迁移过程，几乎实时地捕获来自 Oracle 源数据库的更改并将其复制到目标的 Aurora PostgreSQL 兼容版或 RDS for PostgreSQL 数据库，确保了数据在两个平台之间的平滑转换并进行转换和映射。

解决方案概述

下图展示了将存储在 Oracle CLOB 数据类型列中的 JSON 数据迁移到 Aurora PostgreSQL 兼容版或 RDS for PostgreSQL JSON 或 JSONB 数据类型列的总体解决方案架构。

在此场景中，我们假设应用程序或用户将大量记录填充到 Oracle 表中。我们在 Amazon Aurora PostgreSQL 兼容版或 Amazon RDS for PostgreSQL 中创建一个暂存表，其中 Oracle 的 CLOB 列被 PostgreSQL 的 TEXT 列替换。AWS DMS 将数据从源迁移到暂存表，PL/pgSQL 程序进一步迁移数据，从暂存表到目标表，捕获无效 JSON 记录。初始加载完成后并且 CDC 正在进行时，依据增量数据大小的调度程序执行此过程。

为实施该解决方案，将完成以下步骤：

前提条件

为了成功地从 Oracle 的 CLOB 迁移到 Aurora PostgreSQL 兼容版和 RDS for PostgreSQL 的 JSON 或 JSONB 列，需满足以下关于 AWS DMS 设置和源/目标数据库配置的前提条件。

AWS DMS

确保您已设置 AWS DMS，以满足以下条件，以便将数据从源表转移到暂存表：

源数据库

源数据库应拥有符合以下要求的表对象，以进行数据传输：

目标数据库

目标数据库应具备满足以下要求的表对象，以接收来自 AWS DMS 的数据，并运行 PL/pgSQL 程序以捕获无效的 JSON 记录：

sql CREATE TABLE discardrecordlog ( tbname text rangestart numeric rangeend numeric failuremessage text discardrecidpkval numeric procexeccounter numeric )

CREATE TABLE exechistoryoffset ( tbname varchar procexeccounter numeric rangeinterval numeric numranges numeric maxrecidatendofiteration numeric )

限制

该解决方案帮助识别在从 Oracle 数据库迁移到 Amazon Aurora PostgreSQL 兼容版或 RDS for PostgreSQL 数据库时存储在 CLOB 列中的无效 JSON 记录。然而，所用的程序主要专注于从暂存表基于主键复制数据到目标表。请注意以下限制：

设置 AWS DMS

在以下步骤中，我们将使用 Amazon Aurora PostgreSQL 兼容版，但同样的解决方案也适用于 Amazon RDS for PostgreSQL。完成以下步骤以设置 AWS DMS，创建 AWS DMS 任务针对源和目标，创建暂存表，运行 AWS DMS 任务并迁移 JSON 数据：

在 AWS DMS 任务创建过程中，进行 创建表映射，用 Oracle 中的 CLOB 列替换暂存表的 TEXT 列。 根据前述 AWS DMS 任务的代码片段，我们有三个源表：SOURCETABLE1、SOURCETABLE2 和 SOURCETABLE3。对于每个表，在 AWS DMS 中配置规则以复制到暂存表。

启动 AWS DMS 任务，将位于本地的 Oracle JSON 数据从 CLOB 列迁移至 Amazon Aurora PostgreSQLCompatible Edition 的 JSON 数据在 TEXT 列中。如果 AWS DMS 任务遇到任何问题，请参考 AWS DMS 故障排除指南。

生成无效 JSON 于源数据库

在实际环境中，无效的 JSON 数据通常通过应用程序接收。为了本文的展示，通过在源 Oracle 数据库中使用以下查询来模拟无效 JSON 记录：

示例 1 插入 jsoncolumn1 和 jsoncolumn2 的有效 JSON 记录：

sqlINSERT INTO SOURCETABLE1 (column1 column2 jsoncolumn1 jsoncolumn2)VALUES (1 Value2 {key1 value1 key2 value2} {key3 value3 key4 value4}) (3 Value4 {key5 value5 key6 value6} {key7 value7 key8 value8}) (5 Value6 {key9 value9 key10 value10} {key11 value11 key12 value12})

sqlINSERT INTO SOURCETABLE1 (column1 column2 jsoncolumn1 jsoncolumn2)VALUES (1 value2 {invalidjson} {invalidjson value}) (3 value4 {invalidjson} {invalidjson value}) (5 value6 {invalidjson} {invalidjson value})

sqlINSERT INTO SOURCETABLE1 (column1 column2 jsoncolumn1 jsoncolumn2)VALUES (1 Value2 {key1 value1 key2 value2} {key1 value1 key2 value2}) (3 Value4 {key1 value1 key2 value2} Invalid JSON Data) (5 Value6 Invalid JSON Data {key1 value1 key2 value2}) (7 Value8 Invalid JSON Data Invalid JSON Data)

创建 PL/pgSQL 程序以捕获无效 JSON 记录

以下程序 jsonvalidationproc 展示了如何将数据从暂存表复制到 Amazon Aurora PostgreSQL 兼容版的目标表，并将无效或跳过的 JSON 记录记录在单独的表中，以便应用管理员或数据库管理员进行验证和修复数据：

sqlCREATE OR REPLACE PROCEDURE jsonvalidationproc( sourcetablename text targettablename text rangeinterval numeric jsoncolumns text [] primarykeycolumn text loadtype text force boolean DEFAULT false)LANGUAGE plpgsqlAS DECLARE nonjsoncolumns text jsoncolumnsstr text rangestart numeric = 1 rangeend numeric rangeoffsetleft numeric rowcount numeric insertquery text columncasts text[] temppkrecord RECORD procexeccounter numeric preexecmaxpkid numeric offsetvalue numeric targettablerowcheck numericBEGIN IF loadtype = il THEN 如果不存在则创建 discardrecordlog 和 exechistoryoffset 表 EXECUTE CREATE TABLE IF NOT EXISTS discardrecordlog ( tbname text rangestart numeric rangeend numeric failuremessage text discardrecidpkval numeric procexeccounter numeric ) EXECUTE CREATE TABLE IF NOT EXISTS exechistoryoffset ( tbname text procexeccounter numeric rangeinterval numeric numranges numeric maxrecidatendofprocexeccounter numeric ) ELSIF loadtype = di THEN RAISE NOTICE 您已选择另一次迭代以处理增量插入，之前的初始或增量插入 ELSE RAISE EXCEPTION 无效的加载类型 loadtype END IF

 生成查询以获取 exechistoryoffset 表中最大的 procexeccounterEXECUTE SELECT COALESCE(MAX(procexeccounter) 0) FROM exechistoryoffset WHERE tbname = 1 INTO procexeccounter USING sourcetablename 为当前执行递增 procexeccounterprocexeccounter = procexeccounter  1 获取指定源表的最大主键值，以便每次程序执行IF procexeccounter gt 1 THEN            EXECUTE SELECT maxrecidatendofprocexeccounter FROM exechistoryoffset WHERE procexeccounter = (SELECT max(procexeccounter) FROM exechistoryoffset WHERE tbname = 1) and tbname = 2 INTO preexecmaxpkid USING sourcetablename sourcetablenameELSE    preexecmaxpkid = 0     检查目标表的行数，如果初始加载首次失败，则它不会在 procexeccounter 中做记录。    EXECUTE format(SELECT COUNT(1) FROM I targettablename) INTO targettablerowcheckEND IFIF loadtype = il THEN     IF procexeccounter gt 1 OR (procexeccounter = 1 AND targettablerowcheck gt 1) THEN        IF force = true THEN             请求的初始加载，使用强制选项以删除所有先前的插入数据            EXECUTE format(DELETE FROM I targettablename)            EXECUTE DELETE FROM discardrecordlog WHERE tbname = 1 USING sourcetablename            EXECUTE DELETE FROM exechistoryoffset WHERE tbname = 1 USING sourcetablename            COMMIT            preexecmaxpkid = 0            procexeccounter = 1        ELSE            IF procexeccounter gt 1 THEN                RAISE EXCEPTION 您已选择初始加载，但初始加载在主键值 s 之前已经发生，请验证并使用 FORCE 选项为真。如果您错误地执行初始加载，请运行增量插入 preexecmaxpkid            ELSE                RAISE EXCEPTION 您已选择初始加载，但您的前一次初始加载似乎失败，已经有些记录加载至目标，直到初始加载完成标记将无法为增量插入设置，您必须将 FORCE 选项设置为真            END IF        END IF    END IFELSIF loadtype = di THEN    IF procexeccounter = 1 AND targettablerowcheck gt 1 THEN        RAISE EXCEPTION 您选择了增量迁移，但前一次初始加载似乎失败，已存在记录加载至目标，直到初始加载完成标记无法设置，您必须将 FORCE 选项设置为真    ELSIF procexeccounter = 1 THEN        RAISE NOTICE 您选择了增量插入，但尚未运行初始加载，因此这将被视为初始加载    END IFEND IF 生成目标表的列名选择查询，排除 JSON 列EXECUTE SELECT stringagg(columnname  ) FROM informationschemacolumns WHERE tablename = 1 AND columnname NOT IN (SELECT unnest(2text[])) INTO nonjsoncolumns USING targettablename jsoncolumns 生成 JSON 列名的英文逗号分隔字符串jsoncolumnsstr = arraytostring(jsoncolumns  ) 选择查询以获取每个 JSON 列的格式化