通过

使用 Oracle GoldenGate 将 Oracle CDC 数据流式传输到 Fabric Eventstream Kafka 终结点

在本教程中，您将学习如何使用 Oracle GoldenGate（OGG）从 Oracle 数据库中提取和复制 CDC（变更数据捕获）数据，并通过事件流的自定义终结点源提供的 Kafka 终结点，将这些数据复制到 Microsoft Fabric Real-Time Intelligence。 此设置允许对 Oracle CDC 数据进行实时处理，并允许将其发送到 Fabric 中的各种目标，例如 Eventhouse、Reflex、派生流或自定义终结点目标，以便进一步分析。

在本教程中，你将：

  • 创建 Oracle VM 和 Oracle 数据库。
  • 在 Oracle VM 上安装 Oracle GoldenGate （OGG） 核心。
  • 为 OGG 和其提取进程（Extract）配置 Oracle 数据库，以提取 CDC 数据。
  • 在 Oracle VM 上安装 Oracle GoldenGate （OGG） 大数据。
  • 将 OGG 大数据配置为将 CDC 数据复制到 Eventstream 的 Kafka 终结点。
  • 验证从 Oracle 到 Eventstream 的整个端到端流。

先决条件

  • 获取使用“参与者”或更高权限访问事件流所在的工作区的权限。
  • 首选 shell：Windows、Linux 或 Azure Shell。
  • 熟悉 Unix 编辑器（如 vi 或 vim）以及对 X Server 的基本了解。

创建 Oracle VM 和数据库

本部分提供有关使用 Azure CLI 命令创建 Oracle 虚拟机并在其中设置数据库的说明。

创建 Oracle 虚拟机

  1. 打开首选 shell，使用 az login 命令登录到 Azure 订阅。 然后按照屏幕上的说明操作：

    $ az login

  2. 检查使用的订阅名称和 ID，以确保其正确：

    $ az account show

  3. 创建用于对本教程的所有 Azure 资源进行分组的资源组：

    $ az group create --name esoggcdcrg --location eastus2

  4. 创建本教程所需的网络资源：

    1. 在本教程中创建用于虚拟机的虚拟网络（虚拟网络）：
      $ az network vnet create --name oggVnet --resource-group esoggcdcrg --address-prefixes "10.0.0.0/16" --subnet-name oggSubnet1 --subnet-prefixes "10.0.0.0/24"
    2. 创建网络安全组（NSG）：
      $ az network nsg create --resource-group esoggcdcrg --name oggVnetNSG
    3. 创建 NSG 规则以允许虚拟网络中的网络流量：
      $ az network nsg rule create --resource-group esoggcdcrg --nsg-name oggVnetNSG --name oggAllowVnet --protocol '*' --direction inbound --priority 3400 --source-address-prefix 'VirtualNetwork' --source-port-range '*' --destination-address-prefix 'VirtualNetwork' --destination-port-range '*' --access allow
    4. 创建 NSG 规则以允许 RDP 连接连接 Windows VM。 它用于从本地 Windows 计算机远程访问 Windows VM：
      $ az network nsg rule create --resource-group esoggcdcrg --nsg-name oggAllowRDP --protocol '*' --direction inbound --priority 3410 --source-address-prefix '*' --source-port-range '*' --destination-address-prefix '*' --destination-port-range '3389' --access allow
    5. 创建 NSG 规则以拒绝所有入站连接：
      $ az network nsg rule create --resource-group esoggcdcrg --nsg-name oggDenyAllInBound --protocol '*' --direction inbound --priority 3500 --source-address-prefix '*' --source-port-range '*' --destination-address-prefix '*' --destination-port-range '*' --access deny
    6. 将 NSG 分配到托管服务器的子网：
      $ az network vnet subnet update --resource-group esoggcdcrg --vnet-name oggVNet --name oggSubnet1 --network-security-group oggVnetNSG

  5. 创建 Oracle 虚拟机：

    $ az vm create --resource-group esoggcdcrg --name oggVM --image Oracle:oracle-database:oracle_db_21:21.0.0 --size Standard_D2s_v3 --generate-ssh-keys --admin-username azureuser --vnet-name oggVnet --subnet oggSubnet1 --public-ip-address "" --nsg "" --zone 1

创建和配置 X Server VM

若要创建 Oracle 数据库，需要使用 SSH 登录到在上一步中创建的虚拟机。 由于之前定义的 NSG 规则拒绝了所有入站连接，但允许在虚拟网络内建立连接并允许通过 RDP 3389 端口访问，因此创建了一个 Windows 虚拟机以连接到 Oracle 虚拟机。 此 Windows 虚拟机还用于托管 X 服务器，该服务器将在稍后在 Oracle VM 上安装 GoldenGate 核心应用程序时接收图形安装接口。

  1. 替换密码并运行以下命令以创建部署 X Server 的 Windows 工作站 VM。

    $ az vm create --resource-group esoggcdcrg --name oggXServer --image MicrosoftWindowsDesktop:windows-11:win11-24h2-pro:26100.2605.241207 --size Standard_D2s_v3 --vnet-name oggVnet --subnet oggSubnet1 --public-ip-sku Standard --nsg "" --data-disk-delete-option Delete --os-disk-delete-option Delete --nic-delete-option Delete --security-type Standard --admin-username azureuser --admin-password <YOUR_OWN_PASSWORD>

  2. 创建后，转到 Azure 门户，从 oggXServer 资源下载 RDP 文件，以便使用它来远程访问此 Windows 计算机。 打开 RDP 文件，并使用在创建 Windows VM 时指定的密码（oggXServer）。

    显示如何创建 Windows VM 的屏幕截图。

  3. 在 Windows VM（oggXServer）上安装 WSL，以便 SSH 可用于连接到 Oracle VM（oggVM）。 还可以使用其他 SSH 工具（如 putty）进行连接。

    PS C:\Users\azureuser> wsl --install -d Ubuntu

    显示如何在 Windows VM 上安装 WSL 的屏幕截图。

  4. 需要从 oggVM 下载用于 SSH 登录 oggXServer 的私钥。 下载密钥后，使用此密钥（将此密钥移动到 WSL 主页 .ssh 目录）登录：

    $ ssh -i ~/.ssh/id_rsa.pem azureuser@10.0.0.4

    显示如何更新 ssh 密钥的屏幕截图。

  5. 使用 SSH 将 Oracle VM 与其专用 IP 地址连接。

    显示如何获取专用 IP 地址的屏幕截图。

创建 Oracle 数据库

现在，应该能够使用 SSH 登录到 X 服务器 Windows VM（oggXServer）上的 Oracle VM。 按照以下步骤创建 Oracle 数据库。

  1. 使用 SSH 登录到 Oracle VM（oggVM）。

    $ chmod 400 ~/.ssh/oggVM.pem
$ ssh -i ~/.ssh/oggVM.pem azureuser@10.0.0.4

  2. 将用户更改为 oracle。

    $ sudo su - oracle

  3. 使用以下命令创建数据库。 此命令可能需要 30-40 分钟才能完成。

    $ dbca -silent -createDatabase -templateName General_Purpose.dbc -gdbname cdb1 -sid cdb1 -responseFile NO_VALUE -characterSet AL32UTF8 -sysPassword OraPasswd1 -systemPassword OraPasswd1 -createAsContainerDatabase true -numberOfPDBs 1 -pdbName pdb1 -pdbAdminPassword OraPasswd1 -databaseType MULTIPURPOSE -automaticMemoryManagement false -storageType FS -datafileDestination "/u01/app/oracle/oradata/" -ignorePreReqs

  4. 设置ORACLE_SID和LD_LIBRARY_PATH变量

    $ export ORACLE_SID=cdb1
$ export LD_LIBRARY_PATH=$ORACLE_HOME/lib

  5. 将ORACLE_SID和LD_LIBRARY_PATH添加到 ~/.bashrc 文件，以便保存这些设置以供将来登录。 ORACLE_HOME变量应已在 .bashrc 文件中设置

    $ sed -i '$ a export ORACLE_SID=cdb1' .bashrc
$ sed -i '$ a export LD_LIBRARY_PATH=$ORACLE_HOME/lib' .bashrc

  6. 启动 Oracle 侦听器

    $ lsnrctl start

现在，已创建 Oracle 数据库。 若要启用 Oracle CDC for GoldenGate，需要启用存档日志。 按照以下步骤启用它。

  1. 连接到 sqlplus：

    $ sqlplus / as sysdba

  2. 使用以下命令启用存档日志。 逐个执行以下命令：

    SQL> SELECT log_mode FROM v$database;
SQL> SHUTDOWN IMMEDIATE;
SQL> STARTUP MOUNT;
SQL> ALTER DATABASE ARCHIVELOG;
SQL> ALTER DATABASE OPEN;

  3. 启用强制日志记录和 GoldenGate 复制，并确保至少有一个日志文件存在。 逐个执行以下命令：

    SQL> ALTER DATABASE FORCE LOGGING;
SQL> ALTER SYSTEM SWITCH LOGFILE;
SQL> ALTER SYSTEM set enable_goldengate_replication=true;
SQL> ALTER PLUGGABLE DATABASE PDB1 OPEN;
SQL> ALTER SESSION SET CONTAINER=CDB$ROOT;
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
SQL> ALTER SESSION SET CONTAINER=PDB1;
SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
SQL> EXIT;

安装 Oracle GoldenGate Core

本部分介绍如何下载 Oracle GoldenGate Core 应用程序并将其传输到 Oracle VM（oggVM），并安装它。 所有这些步骤都在 X 服务器 Windows VM（oggXServer）上执行。

将 Oracle GoldenGate Core 应用程序下载并传输到 VM

  1. 打开 X 服务器 Windows VM（oggXServer）的 RDP，然后在 Windows VM（oggXServer）上从 Oracle Golden Gate 下载 Oracle GoldenGate 21.3.0.0.0 for Oracle on Linux x86-64。

  2. 在 X 服务器 Windows VM（oggXServer）上，使用安全复制协议（SCP）将下载的 zip 文件传输到 Oracle VM（oggVM）。

    $ scp -i ~/.ssh/oggVM.pem 213000_fbo_ggs_Linux_x64_Oracle_shiphome.zip azureuser@10.0.0.4:~/

  3. 若要使用 GUI 接口安装 Oracle GoldenGate 核心应用程序，需要在 oggXServer 上安装 Xming 应用程序。 将 Xming X Server for Windows 下载到 ggXServer，并使用所有默认选项进行安装。

    • 确保未在安装结束时选择“启动”
    • 从“开始”菜单启动“XLAUNCH”应用程序。

  4. 通过从“开始”菜单启动 XLAUNCH 应用程序来完成配置。 请确保选择“无访问控制”。

在 oggVM 上安装 Oracle GoldenGate Core 应用程序

本部分中的所有操作都在 Oracle VM（oggVM）上执行。 因此，请使用 SSH 登录到此 VM，并按照以下步骤安装它。

  1. 使用 SSH 连接到 oggVM。

    $ ssh -i ~/.ssh/oggVM.pem azureuser@10.0.0.4

  2. 将上传的 zip 文件移动到 oracle home dir。

    $ sudo su -
$ mv /home/azureuser/213000_fbo_ggs_Linux_x64_Oracle_shiphome.zip /home/oracle/

  3. 解压缩文件（安装解压缩实用工具（如果尚未安装）。

    $ yum install unzip
$ cd /home/oracle/
$ unzip 213000_fbo_ggs_Linux_x64_Oracle_shiphome.zip

  4. 更改权限。

    $ chown -R oracle:oinstall fbo_ggs_Linux_x64_Oracle_shiphome/
$ exit

  5. 现在，让我们开始安装 Oracle GoldenGate Core。 “DISPLAY=10.0.0.5：0.0”中的专用 IP 是 oggXServer 的专用 IP。

    $ sudo su - oracle
$ export DISPLAY=10.0.0.5:0.0
$ cd fbo_ggs_Linux_x64_Oracle_shiphome/Disk1
$ ./runInstaller

    此时应显示 Xming 服务器已由此安装程序打开。

  6. 选择“Oracle GoldenGate for Oracle Database 21c”。 然后选择 下一步 以继续。

  7. 选择软件安装目录为 /u01/app/oggcore，确保选中 启动管理器 框，然后选择 下一步 继续。

  8. 在摘要步骤中选择 安装。

  9. 在最后一步中选择“关闭”。

现在，Oracle GoldenGate 核心应用程序已成功安装在 Oracle VM（oggVM）中。

为 OGG 和 OGG Extract 配置 Oracle 数据库以提取 CDC 数据

安装 Oracle GoldenGate 核心应用程序后，可以将其配置为提取 Oracle CDC 数据。 请按照以下步骤配置“提取”。 本部分中的所有操作仍通过 SSH 连接在 Oracle VM（oggVM）上执行。

准备数据库以用于提取

  1. 创建或更新 tnsnames.ora 文件。

    $ sudo su - oracle
$ cd $ORACLE_HOME/network/admin
$ vi tnsnames.ora

    打开 vi 编辑器时，必须按 i 切换到插入模式，然后复制并粘贴文件内容，然后按 Esc 键，:wq 保存文件。

    cdb1=
 (DESCRIPTION=
   (ADDRESS=
     (PROTOCOL=TCP)
     (HOST=localhost)
     (PORT=1521)
   )
   (CONNECT_DATA=
     (SERVER=dedicated)
     (SERVICE_NAME=cdb1)
   )
 )

pdb1=
 (DESCRIPTION=
   (ADDRESS=
     (PROTOCOL=TCP)
     (HOST=localhost)
     (PORT=1521)
   )
   (CONNECT_DATA=
     (SERVER=dedicated)
     (SERVICE_NAME=pdb1)
   )
 )

  2. 创建 Golden Gate 所有者和用户帐户。

    $ sqlplus / as sysdba

    SQL> CREATE USER C##GGADMIN identified by ggadmin;
SQL> EXEC dbms_goldengate_auth.grant_admin_privilege('C##GGADMIN',container=>'ALL');
SQL> GRANT DBA to C##GGADMIN container=all;
SQL> connect C##GGADMIN/ggadmin
SQL> ALTER SESSION SET CONTAINER=PDB1;
SQL> EXIT;

  3. 创建 Golden Gate ggtest 用户帐户。

    $ cd /u01/app/oggcore
$ sqlplus system/OraPasswd1@pdb1

    SQL> CREATE USER ggtest identified by ggtest DEFAULT TABLESPACE USERS TEMPORARY TABLESPACE TEMP;
SQL> GRANT connect, resource, dba TO ggtest;
SQL> ALTER USER ggtest QUOTA 100M on USERS;
SQL> connect ggtest/ggtest@pdb1
SQL> @demo_ora_create
SQL> @demo_ora_insert
SQL> select * from TCUSTMER;
SQL> select * from TCUSTORD;
SQL> EXIT;

应能够看到两个表（TCUSTMER 和 TCUSTORD）已创建，并在两个表中插入两条记录。

配置并启用提取

  1. 为 Oracle GoldenGate Extract 配置提取参数文件。

    $ sudo su – oracle

  2. 确保在 ~/.bashrc中设置 $TNS_ADMIN。 否则，请将其设置为：~/.bashrc中的 export TNS_ADMIN=$ORACLE_HOME/network/admin。 并运行命令 source ~/.bashrc 使其生效。

    $ cd /u01/app/oggcore
$ ./ggsci
GGSCI> DBLOGIN USERID ggtest@pdb1

Successfully logged into database  pdb1

GGSCI> ADD SCHEMATRANDATA pdb1.ggtest

2025-01-04 15:57:42  INFO    OGG-01788  SCHEMATRANDATA has been added on schema "ggtest".
2025-01-04 15:57:42  INFO    OGG-01976  SCHEMATRANDATA for scheduling columns has been added on schema "ggtest".
2025-01-04 15:57:42  INFO    OGG-10154  Schema level PREPARECSN set to mode NOWAIT on schema "ggtest".

GGSCI> EDIT PARAMS EXT1

    EXTRACT EXT1
USERID C##GGADMIN@cdb1, PASSWORD ggadmin
RMTHOST 10.0.0.4, MGRPORT 7809
RMTTRAIL ./dirdat/rt
DDL INCLUDE MAPPED
DDLOPTIONS REPORT
LOGALLSUPCOLS
UPDATERECORDFORMAT COMPACT
TABLE pdb1.ggtest.TCUSTMER;
TABLE pdb1.ggtest.TCUSTORD;

  3. 注册提取 - 集成提取。

    GGSCI> dblogin userid C##GGADMIN@cdb1

Successfully logged into database CDB$ROOT.

GGSCI> REGISTER EXTRACT EXT1 DATABASE CONTAINER(pdb1)

2025-01-04 16:04:58  INFO    OGG-02003  Extract group EXT1 successfully registered with database at SCN 2147164.

GGSCI> exit

  4. 设置提取检查点并启动实时提取。

    GGSCI> ADD EXTRACT EXT1, INTEGRATED TRANLOG, BEGIN NOW
GGSCI> ADD RMTTRAIL ./dirdat/rt, EXTRACT EXT1, MEGABYTES 10

RMTTRAIL added.

GGSCI> START EXTRACT EXT1

Sending START request to MANAGER ...
EXTRACT EXT1 starting

GGSCI> INFO ALL

Program     Status      Group       Lag at Chkpt  Time Since Chkpt
MANAGER     RUNNING
EXTRACT     RUNNING     EXTORA      00:00:11      00:00:04

GGSCI > EXIT

  5. 对配置的提取进行冒烟测试。

    1. 使用测试帐户登录 DB，并将记录插入表：

      $ sqlplus ggtest

      SQL> select * from TCUSTORD;
SQL> INSERT INTO TCUSTORD VALUES ('OLLJ',TO_DATE('11-JAN-25'),'CYCLE',400,16000,1,102);
SQL> COMMIT;
SQL> EXIT;

    2. 检查 Golden Gate 获取的事务（记下“插入总数”值）：

      $ cd /u01/app/oggcore
$ ./ggsci
GGSCI> STATS EXT1

Sending STATS request to Extract group EXT1 ...
Start of statistics at 2025-01-04 16:12:16.
DDL replication statistics (for all trails):

*** Total statistics since extract started     ***
    Operations                                         0.00
    Mapped operations                                  0.00
    Unmapped operations                                0.00
    Other operations                                   0.00
    Excluded operations                                0.00

Output to ./dirdat/rt:
Extracting from PDB1.GGTEST.TCUSTORD to PDB1.GGTEST.TCUSTORD:
*** Total statistics since 2025-01-04 16:11:02 ***
    Total inserts                              1.00
    Total updates                              0.00
    Total deletes                              0.00
    Total upserts                              0.00
    Total discards                             0.00
    Total operations                           1.00

*** Daily statistics since 2025-01-04 16:11:02 ***
    Total inserts                              1.00
    Total updates                              0.00
    Total deletes                              0.00
    Total upserts                              0.00
    Total discards                             0.00
    Total operations                           1.00

*** Hourly statistics since 2025-01-04 16:11:02 ***
    Total inserts                              1.00
    Total updates                              0.00
    Total deletes                              0.00
    Total upserts                              0.00
    Total discards                             0.00
    Total operations                           1.00

*** Latest statistics since 2025-01-04 16:11:02 ***
    Total inserts                              1.00
    Total updates                              0.00
    Total deletes                              0.00
    Total upserts                              0.00
    Total discards                             0.00
    Total operations                           1.00
End of statistics.

安装 Oracle GoldenGate Big Data

Oracle GoldenGate 大数据将提取的数据复制到所需的处理程序（目标）。 在本教程中，处理程序是 Kafka 主题。 在本部分中，将指导你下载所需的软件包并安装它们。

下载所需的软件包

Windows VM（oggXServer）仍用于下载这些软件包并将其传输到 Oracle VM（oggVM）。

  1. 从 Oracle GoldenGate Downloads 下载适用于大数据的 OGG（适用于 Linux x86-64 上的 Oracle GoldenGate for Big Data 21.4.0.0.0）。

  2. 下载 Kafka 包（kafka_2.13-3.9.0.tgz）。

  3. 下载 JAVA SDK(jdk-8u202-linux-x64.tar.gz。

  4. 使用 WSL 中的 SCP 命令将其传输到 Oracle VM：

    $ scp -i ~/.ssh/oggVM.pem kafka_2.13-3.9.0.tgz jdk-8u202-linux-x64.tar.gz 214000_ggs_Linux_x64_BigData_64bit.zip azureuser@10.0.0.4:~/

安装三个软件包

若要安装这三个软件包，请将其提取到各个文件夹，并配置相应的环境变量。 本部分中的所有操作都在 Oracle VM（oggVM）上执行。 因此，请使用 SSH 登录到此 VM，并按照以下步骤安装这些 VM。

  1. 使用 SSH 连接到 oggVM。

    $ ssh -i ~/.ssh/oggVM.pem azureuser@10.0.0.4

  2. 将这些包移动到 oracle home dir。

    $ sudo su –
$ mv /home/azureuser/214000_ggs_Linux_x64_BigData_64bit.zip /home/azureuser/kafka_2.13-3.9.0.tgz /home/azureuser/jdk-8u202-linux-x64.tar.gz /home/oracle/
$ exit

  3. 更改为 oracle 用户并转到主目录以准备各个文件夹。

    $ sudo su – oracle
$ mkdir kafka java oggbd
$ mv 214000_ggs_Linux_x64_BigData_64bit.zip oggbd/
$ mv jdk-8u202-linux-x64.tar.gz java/
$ mv kafka_2.13-3.9.0.tgz kafka

  4. 转到每个单独的文件夹并提取其中的包。

    $ cd java
$ tar -xvf jdk-8u202-linux-x64.tar.gz
$ cd ../kafka
$ tar -xvf kafka_2.13-3.9.0.tgz
$ cd ../oggbd
$ mkdir /u01/app/oggbd
$ tar -xvf ggs_Linux_x64_BigData_64bit.tar -C /u01/app/oggbd

  5. 为 JAVA 配置环境变量。

    $ export JAVA_HOME=/home/oracle/java/jdk1.8.0_202
$ export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$JAVA_HOME/lib:/$JAVA_HOME/jre/lib/amd64/server

将 Oracle GoldenGate 大数据配置为将 CDC 数据复制到 Eventstream 的 Kafka 终结点

在本部分中，将引导你配置 Oracle GoldenGate 大数据，以便将上一部分中提取的 CDC 数据复制到 Eventstream 的 Kafka 终结点。

设置 Eventstream Kafka 端点

按照 Eventstream 及其自定义终结点源的正常创建过程获取 Kafka 终结点信息以供以后使用，请参阅 将自定义终结点或自定义应用源添加到事件流。

显示已成功创建自定义终结点源的屏幕截图。

为 Oracle GoldenGate Big Data 配置复制

本部分中的所有操作都在 Oracle VM（oggVM）上执行。 因此，请使用 SSH 登录到此 VM，并按照以下步骤对其进行配置。

  1. 如果在上一部分后丢失了连接，请使用 SSH 连接到 oggVM。

    $ ssh -i ~/.ssh/oggVM.pem azureuser@10.0.0.4
$ sudo su - oracle

  2. 转到将 Oracle GoldenGate Big Data 包提取到的文件夹并创建所需文件夹。

    $ cd /u01/app/oggbd
$ ./ggsci
GGSCI> CREATE SUBDIRS
GGSCI> EDIT PARAMS MGR

    PORT 7801

    GGSCI> START MGR
GGSCI> INFO ALL

Program     Status      Group       Lag at Chkpt  Time Since Chkpt
MANAGER     RUNNING

GGSCI> EXIT

  3. 复制 Kafka 处理程序模板配置文件。

    $ cp AdapterExamples/big-data/kafka/* dirprm/

  4. 打开 rkafka.prm 文件，并根据源数据库中的架构名称更改 MAP/TARGET 架构名称。

    $ vim dirprm/rkafka.prm

    REPLICAT rkafka
TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 10000
MAP pdb1.ggtest.*, TARGET pdb1.ggtest.*;

  5. 在 ./ggsci 命令中使用以下命令添加复制。 确保复制应为提取步骤中定义的复制：

    $ ./ggsci
GGSCI> ADD REPLICAT rkafka, exttrail /u01/app/oggcore/dirdat/rt
GGSCI> EXIT

  6. 打开 Kafka props 文件，并将 gg.classpath 更改为 apache Kafka #Sample gg.classpath 下的正确 kafka 安装目录路径。 此外，请填写来自 Eventstream 自定义终结点源的 Kafka 主题名称。

    $ vim dirprm/kafka.props

    gg.handlerlist=kafkahandler
gg.handler.kafkahandler.type=kafka
gg.handler.kafkahandler.kafkaProducerConfigFile=custom_kafka_producer.properties
#The following resolves the topic name using the fixed topic which is from eventstream
gg.handler.kafkahandler.topicMappingTemplate={YOUR.TOPIC.NAME}
#The following selects the message key using the concatenated primary keys
#A null Kafka message key distributes to the partitions on a round-robin basis
gg.handler.kafkahandler.keyMappingTemplate=${null}
#gg.handler.kafkahandler.schemaTopicName=mySchemaTopic
gg.handler.kafkahandler.blockingSend=false
gg.handler.kafkahandler.includeTokens=false
gg.handler.kafkahandler.mode=op
#gg.handler.kafkahandler.metaHeadersTemplate=${alltokens}
gg.handler.kafkahandler.transactionsEnabled=false
gg.handler.kafkahandler.format=json
gg.handler.kafkahandler.format.metaColumnsTemplate=${objectname[table]},${optype[op_type]},${timestamp[op_ts]},${currenttimestamp[current_ts]},${position[pos]}
#Sample gg.classpath for Apache Kafka
gg.classpath=dirprm/:/var/lib/kafka/libs/*:/home/oracle/kafka/kafka_2.13-3.9.0/libs/*
#Sample gg.classpath for HDP
#gg.classpath=/etc/kafka/conf:/usr/hdp/current/kafka-broker/libs/*
    • 可以在“Kafka”选项卡下的“SAS 密钥身份验证” 页上找到 {YOUR.TOPIC.NAME} 值：

    显示如何获取主题名称的屏幕截图。

  7. 通过添加连接到 Eventstream 所需的 Eventstream 连接字符串和密码，对 custom_producer.properties 文件进行更改。

    $ vim dirprm/custom_kafka_producer.properties

    bootstrap.servers={YOUR.BOOTSTRAP.SERVER}
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config={YOUR.SASL.JASS.CONFIG};

acks=1
reconnect.backoff.ms=1000

value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
# 100KB per partition
batch.size=16384
linger.ms=0
    • 将 {YOUR.BOOTSTRAP.SERVER} 替换为“Bootstrap 服务器”值，可从你的事件流中复制此值。
    • 将 {YOUR.SASL.JASS.CONFIG} 替换为“SASL JASS 配置”值，可从你的事件流中复制此值。

    显示如何获取启动和连接密钥的屏幕截图。

  8. 登录到 GoldenGate 实例并启动复制过程。

    $ ./ggsci
GGSCI> START RKAFKA

Sending START request to Manager ...
Replicat group RKAFKA starting.

    GGSCI> INFO ALL
Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING
REPLICAT    RUNNING     RKAFKA      00:00:00      00:53:17

GGSCI> EXIT

验证从 Oracle 到 Eventstream 的整个 E2E 流

若要验证整个 E2E 流，让我们使用 ggtest 帐户登录到 Oracle 数据库，以插入一些记录，然后转到 Eventstream 检查更改数据流是否传入。

  1. 使用测试帐户登录 Oracle DB 以插入一些新记录：

    $ sqlplus ggtest

    SQL> select * from TCUSTORD;
SQL> INSERT INTO TCUSTORD VALUES ('MSFT',TO_DATE('3-JAN-25'),'M365',100,80000,2,104);
SQL> INSERT INTO TCUSTMER VALUES ('TOM','SUNRISE SOFTWARE INC.','SAN FRANCISCO','CA');
SQL> COMMIT;
SQL> EXIT;

  2. 预览使用此 Kafka 终结点源发送的数据。 选择默认流节点，该节点是显示事件流名称的中间节点。

    显示如何在事件流中预览数据的屏幕截图。