从 Kafka 平台捕获数据

使用 OCI GoldenGate 从 Kafka 平台流处理源中提取消息。

概述

您可以使用 OCI GoldenGate 从以下流处理源捕获消息：

• Apache Kafka

• OCI 流处理

• Confluent Kafka，带或不带 Confluent Schema Registry

• Azure 事件中心

• Amazon MSK

• Microsoft Fabric 事件流

OCI GoldenGate 从 Kafka 主题或主题读取消息，然后将数据转换为写入 GoldenGate Trail 文件的逻辑更改记录。然后，GoldenGate Replicat 进程可以使用生成线索文件来传播数据以支持 RDBMS 实施。

任务 1：配置使用者属性

1. 使用以下反序列化器或转换器之一创建 Kafka 使用者属性文件。如果源是 Confluent Kafka 与 Confluent Schema Registry 中的一个主题，则可以使用 Avro 转换器。对于其他源，请根据需要使用 JSON 转换器或反序列化器：

   • JSON 反序列化器的 Kafka 使用者属性：

     key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
     value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

   • JSON 转换器的 Kafka 使用者属性：

     key.converter=org.apache.kafka.connect.json.JsonConverter
     value.converter=org.apache.kafka.connect.json.JsonConverter

   • 适用于 Avro 转换器的 Kafka 使用者属性：

     key.converter=io.confluent.connect.avro.AvroConverter
     value.converter=io.confluent.connect.avro.AvroConverter

2. 保存属性文件并记下其位置。

任务 2：创建 OCI GoldenGate 资源

此任务将指导您如何创建新资源（如果尚不存在）。确保您正在使用的大数据部署升级到最新版本。

1. 为大数据创建 OCI GoldenGate 部署。

2. 创建连接。

   注：创建以下任一连接时，请确保选择显示高级选项，然后上载使用者属性文件。

   • 对于 Apache Kafka 或 Amazon MSK，请创建 Kafka 连接。

   • 对于 Confluent Kafka，请创建 Confluent Kafka 连接。

   • 对于 Confluent Schema Registry，请创建 Confluent Schema Registry 连接。

   • 对于 Azure 事件中心，请创建 Azure 事件中心连接。

   • 对于 OCI Streaming，请创建 OCI Streaming 连接。

   • 对于 Microsoft Fabric Eventstream，创建 Microsoft Fabric Eventstream 连接

3. 将连接分配给大数据部署。

任务 3：创建身份证明

必须先创建身份证明，然后才能创建提取。

要创建身份证明，请执行以下操作：

1. 在“Deployments（部署）”页面上选择“Big Data deployment（大数据）”部署。

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

3. 使用在任务 2 步骤 1 中创建部署时指定的用户名和密码登录到大数据部署。

4. 从导航菜单中选择 DB Connections（DB 连接）。

5. 在“配置”页的“数据库”选项卡下，选择添加数据库连接（加号图标），然后按如下方式填写表单：

   1. 对于 Credential Domain（身份证明域），输入 OracleGoldenGate。

   2. 对于身份证明别名，输入 kafka。

   3. 对于用户 ID ，输入 kafka://

   4. 对于 Password 和 Verify Password ，输入口令。

   5. 选择提交。

任务 4：创建提取

1. 在主页上，选择添加提取（加号图标）。

2. 使用以下值添加提取：

   1. 在“提取信息”页面上，按如下方式填写字段，然后选择下一步：

      • 对于提取类型，选择“集成提取”。

      • 在流程名称中，输入提取的名称。

   2. 在“提取选项”页面上，按如下方式填写字段，然后选择下一步：

      • 对于域，选择域。

      • 对于别名，选择分配给部署的连接。

      • 对于名称，输入 2 个字符的名称。

   3. 在“托管选项”页面上，将字段保留原样，然后选择下一步。

   4. 在“参数文件”页上：

      • 将表映射保留为 TABLE TESTSCHEMA.*; 以侦听给定引导服务器中的所有主题。您还可以将表映射设置为 TABLE TESTSCHEMA.<topic-name>; 以从指定的主题捕获。

      • 将 SOURCEDB USERIDALIAS 更新为 SOURCEDB USERIDALIAS kafka DOMAIN OracleGoldenGate。

   5. 选择创建并运行。

返回到 Oracle GoldenGate 主页，您可以在其中查看提取流程启动和查看事件消息。