通过数据验证部署验证数据一致性

了解如何使用 OCI GoldenGate 验证源数据库和目标数据库之间的数据一致性。

开始之前

要成功完成此快速入门，您必须具有：

任务 1：设置的环境

  1. 创建数据复制部署。

  2. 创建源 Oracle Autonomous AI Transaction Processing (ATP) 连接。

  3. 创建目标自治 AI 数据湖仓 (ALK) 连接。

  4. 分配到部署的连接。

  5. 使用 SQL 工具启用补充日志记录：

    ALTER DATABASE ADD SUPPLEMENTAL LOG DATA

  6. 在 SQL 工具中运行以下查询，以确保源数据库中的所有表都使用 support_mode=FULL：

    select * from DBA_GOLDENGATE_SUPPORT_MODE where owner = 'SRC_OCIGGLL';

任务 2：创建集成提取

集成提取捕获对源数据库的持续更改。

  1. 在“部署详细信息”页上，选择启动控制台。

  2. 如果需要，输入 oggadmin 以显示用户名和创建部署时使用的密码，然后选择登录。

  3. 添加事务处理数据和检查点表：

    1. 打开导航菜单，然后选择 DB Connections（DB 连接）。

    2. 选择 Connect to database SourceDB 。

    3. 在导航菜单中，选择 Trandata ，然后选择 Add Trandata （加号图标）。

    4. 对于方案名称，输入 SRC_OCIGGLL，然后选择提交。

    5. 要进行验证，请在“搜索”字段中输入 SRC_OCIGGLL，然后选择搜索。

    6. 打开导航菜单，然后选择 DB Connections（DB 连接）。

    7. 选择 Connect to database TargetDB（连接到数据库 TargetDB）。

    8. 在导航菜单中，选择检查点，然后选择添加检查点（加号图标）。

    9. 对于检查点表，输入 "SRCMIRROR_OCIGGLL"."CHECKTABLE"，然后选择提交。

  4. 添加提取。

    注：有关可用于指定源表的参数的详情，请参阅附加提取参数选项。

    在“提取参数”页上，在 EXTTRAIL <trail-name> 下附加以下行：

    -- Capture DDL operations for listed schema tables
ddl include mapped

-- Add step-by-step history of to the report file. Useful when troubleshooting.
ddloptions report

-- Write capture stats per table to the report file daily.
report at 00:01

-- Rollover the report file weekly. Useful when IE runs
-- without being stopped/started for long periods of time to
-- keep the report files from becoming too large.
reportrollover at 00:01 on Sunday

-- Report total operations captured, and operations per second
-- every 10 minutes.
reportcount every 10 minutes, rate

-- Table list for capture
table SRC_OCIGGLL.*;

  5. 检查长耗时事务处理。在源数据库上运行以下脚本：

    select start_scn, start_time from gv$transaction where start_scn < (select max(start_scn) from dba_capture);

    如果查询返回任何行，则必须找到事务处理的 SCN，然后提交或回退事务处理。

任务 3：使用 Oracle Data Pump 导出数据 (ExpDP)

使用 Oracle Data Pump (ExpDP) 将数据从源数据库导出到 Oracle 对象存储。

  1. 创建 Oracle 对象存储存储桶。

    记下与导出和导入脚本一起使用的名称空间和存储桶名称。

  2. 创建验证令牌，然后将令牌字符串复制并粘贴到文本编辑器以供日后使用。

  3. 在源数据库中创建一个身份证明，将 <user-name> 和 <token> 替换为您的 Oracle Cloud 账户用户名和在上一步中创建的令牌字符串：

    BEGIN
  DBMS_CLOUD.CREATE_CREDENTIAL(
    credential_name => 'ADB_OBJECTSTORE',
    username => '<user-name>',
    password => '<token>'
  );
END;

  4. 在源数据库中运行以下脚本以创建导出数据作业。确保相应地替换对象存储 URI 中的 <region>、<namespace> 和 <bucket-name>。SRC_OCIGGLL.dmp 是将在此脚本运行时创建的文件。

    DECLARE
ind NUMBER;              -- Loop index
h1 NUMBER;               -- Data Pump job handle
percent_done NUMBER;     -- Percentage of job complete
job_state VARCHAR2(30);  -- To keep track of job state
le ku$_LogEntry;         -- For WIP and error messages
js ku$_JobStatus;        -- The job status from get_status
jd ku$_JobDesc;          -- The job description from get_status
sts ku$_Status;          -- The status object returned by get_status

BEGIN

-- Create a (user-named) Data Pump job to do a schema export.
h1 := DBMS_DATAPUMP.OPEN('EXPORT','SCHEMA',NULL,'SRC_OCIGGLL_EXPORT','LATEST');

-- Specify a single dump file for the job (using the handle just returned)
-- and a directory object, which must already be defined and accessible
-- to the user running this procedure.
DBMS_DATAPUMP.ADD_FILE(h1,'https://objectstorage.<region>.oraclecloud.com/n/<namespace>/b/<bucket-name>/o/SRC_OCIGGLL.dmp','ADB_OBJECTSTORE','100MB',DBMS_DATAPUMP.KU$_FILE_TYPE_URIDUMP_FILE,1);

-- A metadata filter is used to specify the schema that will be exported.
DBMS_DATAPUMP.METADATA_FILTER(h1,'SCHEMA_EXPR','IN (''SRC_OCIGGLL'')');

-- Start the job. An exception will be generated if something is not set up properly.
DBMS_DATAPUMP.START_JOB(h1);

-- The export job should now be running. In the following loop, the job
-- is monitored until it completes. In the meantime, progress information is displayed.
percent_done := 0;
job_state := 'UNDEFINED';
while (job_state != 'COMPLETED') and (job_state != 'STOPPED') loop
  dbms_datapump.get_status(h1,dbms_datapump.ku$_status_job_error + dbms_datapump.ku$_status_job_status + dbms_datapump.ku$_status_wip,-1,job_state,sts);
  js := sts.job_status;

-- If the percentage done changed, display the new value.
if js.percent_done != percent_done
then
  dbms_output.put_line('*** Job percent done = ' \|\| to_char(js.percent_done));
  percent_done := js.percent_done;
end if;

-- If any work-in-progress (WIP) or error messages were received for the job, display them.
if (bitand(sts.mask,dbms_datapump.ku$_status_wip) != 0)
then
  le := sts.wip;
else
  if (bitand(sts.mask,dbms_datapump.ku$_status_job_error) != 0)
  then
    le := sts.error;
  else
    le := null;
  end if;
end if;
if le is not null
then
  ind := le.FIRST;
  while ind is not null loop
    dbms_output.put_line(le(ind).LogText);
    ind := le.NEXT(ind);
  end loop;
end if;
  end loop;

  -- Indicate that the job finished and detach from it.
  dbms_output.put_line('Job has completed');
  dbms_output.put_line('Final job state = ' \|\| job_state);
  dbms_datapump.detach(h1);
END;

任务 4：使用 Oracle Data Pump (ImpDP) 实例化目标数据库

使用 Oracle Data Pump (ImpDP) 可以将数据从从源数据库导出的 SRC_OCIGGLL.dmp 导入到目标数据库。

  1. 在目标数据库中创建一个身份证明以访问 Oracle 对象存储（使用前面部分中的相同信息）。

    BEGIN
  DBMS_CLOUD.CREATE_CREDENTIAL(
    credential_name => 'ADB_OBJECTSTORE',
    username => '<user-name>',
    password => '<token>'
  );
END;

  2. 在目标数据库中运行以下脚本以从 SRC_OCIGGLL.dmp 导入数据。确保相应地替换对象存储 URI 中的 <region>、<namespace> 和 <bucket-name>：

    DECLARE
ind NUMBER;  -- Loop index
h1 NUMBER;  -- Data Pump job handle
percent_done NUMBER;  -- Percentage of job complete
job_state VARCHAR2(30);  -- To keep track of job state
le ku$_LogEntry;  -- For WIP and error messages
js ku$_JobStatus;  -- The job status from get_status
jd ku$_JobDesc;  -- The job description from get_status
sts ku$_Status;  -- The status object returned by get_status
BEGIN

-- Create a (user-named) Data Pump job to do a "full" import (everything

-- in the dump file without filtering).
h1 := DBMS_DATAPUMP.OPEN('IMPORT','FULL',NULL,'SRCMIRROR_OCIGGLL_IMPORT');

-- Specify the single dump file for the job (using the handle just returned)
-- and directory object, which must already be defined and accessible
-- to the user running this procedure. This is the dump file created by
-- the export operation in the first example.

DBMS_DATAPUMP.ADD_FILE(h1,'https://objectstorage.<region>.oraclecloud.com/n/<namespace>/b/<bucket-name>/o/SRC_OCIGGLL.dmp','ADB_OBJECTSTORE',null,DBMS_DATAPUMP.KU$_FILE_TYPE_URIDUMP_FILE);

-- A metadata remap will map all schema objects from SRC_OCIGGLL to SRCMIRROR_OCIGGLL.
DBMS_DATAPUMP.METADATA_REMAP(h1,'REMAP_SCHEMA','SRC_OCIGGLL','SRCMIRROR_OCIGGLL');

-- If a table already exists in the destination schema, skip it (leave
-- the preexisting table alone). This is the default, but it does not hurt
-- to specify it explicitly.
DBMS_DATAPUMP.SET_PARAMETER(h1,'TABLE_EXISTS_ACTION','SKIP');

-- Start the job. An exception is returned if something is not set up properly.
DBMS_DATAPUMP.START_JOB(h1);

-- The import job should now be running. In the following loop, the job is

-- monitored until it completes. In the meantime, progress information is

-- displayed. Note: this is identical to the export example.
percent_done := 0;
job_state := 'UNDEFINED';
while (job_state != 'COMPLETED') and (job_state != 'STOPPED') loop
  dbms_datapump.get_status(h1,
    dbms_datapump.ku$_status_job_error +
    dbms_datapump.ku$_status_job_status +
    dbms_datapump.ku$_status_wip,-1,job_state,sts);
    js := sts.job_status;

  -- If the percentage done changed, display the new value.
  if js.percent_done != percent_done
  then
    dbms_output.put_line('*** Job percent done = ' \|\|
    to_char(js.percent_done));
    percent_done := js.percent_done;
  end if;

  -- If any work-in-progress (WIP) or Error messages were received for the job, display them.
  if (bitand(sts.mask,dbms_datapump.ku$_status_wip) != 0)
  then
    le := sts.wip;
  else
    if (bitand(sts.mask,dbms_datapump.ku$_status_job_error) != 0)
    then
      le := sts.error;
    else
      le := null;
    end if;
  end if;
  if le is not null
  then
    ind := le.FIRST;
    while ind is not null loop
      dbms_output.put_line(le(ind).LogText);
      ind := le.NEXT(ind);
    end loop;
  end if;
end loop;

-- Indicate that the job finished and gracefully detach from it.
dbms_output.put_line('Job has completed');
dbms_output.put_line('Final job state = ' \|\| job_state);
dbms_datapump.detach(h1);
END;

任务 5：添加和运行非集成复制

  1. Add and run a Replicat （添加和运行复制）。

    在 Parameter File 屏幕上，将 MAP *.*, TARGET *.*; 替换为以下脚本：

    -- Capture DDL operations for listed schema tables
ddl include mapped

-- Add step-by-step history of ddl operations captured
-- to the report file. Very useful when troubleshooting.
ddloptions report

-- Write capture stats per table to the report file daily.
report at 00:01

-- Rollover the report file weekly. Useful when PR runs
-- without being stopped/started for long periods of time to
-- keep the report files from becoming too large.
reportrollover at 00:01 on Sunday

-- Report total operations captured, and operations per second
-- every 10 minutes.
reportcount every 10 minutes, rate

-- Table map list for apply
DBOPTIONS ENABLE_INSTANTIATION_FILTERING;
MAP SRC_OCIGGLL.*, TARGET SRCMIRROR_OCIGGLL.*;

    注：DBOPTIONS ENABLE_INSTATIATION_FILTERING 允许对使用 Oracle Data Pump 导入的表进行 CSN 筛选。有关更多信息，请参见《 DBOPTIONS Reference 》。

  2. 对源数据库执行插入：

    1. 返回 Oracle Cloud 控制台，然后使用导航菜单导航回 Oracle AI Database 、 Autonomous AI Transaction Processing 和 SourceDB 。

    2. 在 "SourceDB Details"（源数据库详细信息）页面上，选择 Database actions（数据库操作），然后选择 SQL 。

    3. 输入以下插入，然后选择运行脚本：

      Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1000,'Houston',20,743113);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1001,'Dallas',20,822416);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1002,'San Francisco',21,157574);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1003,'Los Angeles',21,743878);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1004,'San Diego',21,840689);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1005,'Chicago',23,616472);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1006,'Memphis',23,580075);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1007,'New York City',22,124434);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1008,'Boston',22,275581);
Insert into SRC_OCIGGLL.SRC_CITY (CITY_ID,CITY,REGION_ID,POPULATION) values (1009,'Washington D.C.',22,688002);

    4. 在 OCI GoldenGate 部署控制台中，选择 Extract name (UAEXT) ，然后选择 Statistics 。验证 SRC_OCIGGLL.SRC_CITY 列出了 10 个插入。

    5. 返回到 "Overview" 屏幕，选择 Replicat name (REP) ，然后选择 Statistics 。确认 SRCMIRROR_OCIGGLL.SRC_CITY 列出了 10 个插入

任务 6：创建数据验证资源

  1. 创建数据验证服务器部署。

  2. 创建 2 个数据验证代理部署。

  3. 创建 GoldenGate 连接。

  4. 将任务 1 中创建的 ATP 连接分配给其中一个代理部署。

  5. 将在任务 1 中创建的 ALK 连接分配给其他代理部署。

  6. 将 GoldenGate 连接分配给服务器部署。

  7. 在部署变为活动状态后，查看数据验证环境：

    1. 在服务器部署的详细信息页面上，选择 Launch console（启动控制台）。

    2. 在导航菜单中，选择连接。

    3. 在“Connections（连接）”页面上，确认您的代理出现在列表中，并且其状态为“Active（有效）”。

任务 7：创建组并比较对

在服务器部署中，创建比较配置：

  1. 在导航菜单中，选择 Groups and Compare Pairs 。

  2. 在“组和比较对”页上，选择创建。

  3. 在“创建组并比较对”配置页上，按如下方式填写字段，然后选择下一步：

    1. 输入名称。

    2. 从其各自的下拉列表中选择源和目标连接。

    3. 从其各自的下拉列表中选择源和目标目录。

    4. 添加比较对

  4. 查看映射规则，然后选择下一步。

  5. 查看映射页，然后选择生成比较对。

  6. 保存组。

任务 8：创建和运行比较作业

  1. 在导航菜单中，选择作业。

  2. 在“作业”页上，选择创建。

  3. 在“创建作业”页上，按如下方式填写字段，然后选择提交：

    1. 输入作业的名称。

    2. 选择添加组，选择在任务 7 中创建的组，然后选择添加并保存。

  4. 在导航菜单中，选择运行作业。

  5. 在“运行作业”页上，选择在步骤 3 中创建的作业和默认概要文件，然后选择运行作业。

任务 9：分析结果

  1. 在导航菜单中，选择监视作业。

  2. 选择已完成的作业。

  3. 选择您的作业以查看其详细信息，然后验证以下内容：

    • 未将任何行标记为不同步。
    • 没有报告缺少的行或额外的行

在此阶段，源数据库和目标数据库已同步。

任务 10：引入数据不一致

在本任务中，您将使用 SRCMIRROR_OCIGGLL.SRC_CITY 表向目标数据库中引入受控不一致。

  1. 验证目标上的现有数据。在目标 ALK SQL 工作表中，输入以下内容：

    SELECT CITY_ID, CITY, REGION_ID, POPULATION
FROM SRCMIRROR_OCIGGLL.SRC_CITY
WHERE CITY_ID BETWEEN 1000 AND 1009
ORDER BY CITY_ID;

  2. 清除工作表，然后输入以下内容以引入数据不匹配：

    UPDATE SRCMIRROR_OCIGGLL.SRC_CITY
SET POPULATION = 999999
WHERE CITY_ID = 1003;

COMMIT;

  3. 清除工作表，然后输入以下内容以引入缺少的行：

    DELETE FROM SRCMIRROR_OCIGGLL.SRC_CITY
WHERE CITY_ID = 1005;

COMMIT;

  4. 清除工作表，然后输入以下内容以引入额外的行：

    INSERT INTO SRCMIRROR_OCIGGLL.SRC_CITY (CITY_ID, CITY, REGION_ID, POPULATION)
VALUES (1010, 'Test City', 99, 123456);

COMMIT;

  5. 清除工作表，然后输入以下内容以验证引入的问题：

    SELECT CITY_ID, CITY, REGION_ID, POPULATION
FROM SRCMIRROR_OCIGGLL.SRC_CITY
WHERE CITY_ID BETWEEN 1000 AND 1010
ORDER BY CITY_ID;

    您应注意以下事项：

    • CITY_ID 1003 的填充不正确
    • 缺少 CITY_ID 1005
    • CITY_ID 1010 仅存在于目标

任务 11：使用 Veridata 修复数据差异

由于源与目标之间存在一些数据差异，因此可以重新运行比较并进行修复。

  1. 在服务器部署控制台导航菜单中，选择运行作业。

  2. 选择“比较”作业，然后选择运行作业。

  3. 在导航菜单中，选择监视作业。

  4. 选择“比较”作业以查看其详细信息。确认以下内容：

    • 行标记为不同步
    • 差异包括：
      • 缺少行
      • 额外行
      • 列不匹配

  5. 在“作业详细信息”页中，找到 SRC_CITY=SRC_CITY，然后选择“比较对”名称以查看其详细信息。

  6. 在“比较对”详细信息中，选择不同步行，然后查看差异。

  7. 选择维修。

  8. 在修复作业对话框中，选择修复所有不同步行，然后选择提交。

    这将创建用于将目标与源同步的修复作业。

  9. 在导航菜单中，选择 Monitor Repair Job 。找到修复作业并对其进行监视，直到其完成。

  10. 修复作业完成后，选择它以查看其详细信息。确认以下内容：

    • 已根据需要更新、插入或删除行。
    • 未发生错误。

  11. 重新运行比较。

    1. 在导航菜单中，选择作业。

    2. 选择“比较”作业，然后选择运行作业。

    3. 导航回监视作业并等待作业完成。

    4. 作业完成后，查看其详细信息以确认以下内容：

      • 没有行标记为不同步
      • 没有缺少的行
      • 无额外行
      • 没有列不匹配

源数据集和目标数据集现在完全同步。