---
title: "创建 OceanBase 数据库至 Kafka 的数据同步项目 | OceanBase 文档中心"
description: 创建 OceanBase 数据库至 Kafka 的数据同步项目 Kafka 是目前广泛应用的高性能分布式流计算平台，OceanBase 迁移服务（OceanBase Migration Service，OMS）支持 OceanBase 两种租户与自建 Kafka 数据源之间的数据实时同步，扩展消息处理能力，广泛应用于…
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*L03BS6f-o40AAAAAAAAAAAAADiGDAQ/original) 迁移服务 OMSV 3.2.2 企业版

# 创建 OceanBase 数据库至 Kafka 的数据同步项目

更新时间：2023-06-25 10:20:37

Kafka 是目前广泛应用的高性能分布式流计算平台，OceanBase 迁移服务（OceanBase Migration Service，OMS）支持 OceanBase 两种租户与自建 Kafka 数据源之间的数据实时同步，扩展消息处理能力，广泛应用于实时数据仓库搭建、数据查询和报表分流等业务场景。

OMS 支持数据同步至消息队列产品，扩展业务在监控数据聚合、流式数据处理、在线和离线分析等大数据领域的全方位应用。

## 使用限制

- 数据同步的对象仅支持物理表，不支持其它对象。
 - 仅支持 0.9 和 2.1 版本的 Kafka 实例。
 - 数据同步过程中，如果您在源端修改了同步范围内的表名称，且重命名后的名称不在同步对象中，则该部分数据将不被同步至目标 Kafka 实例中。
 - 待同步的表名和其中的列名不能包含中文字符。

## 操作步骤

1. 新建同步项目。

      1. 登录 OMS 控制台。
      2. 在左侧导航栏，单击 **数据同步**。
      3. 在 **数据同步** 页面，单击右上角的 **新建同步项目** > **新建数据库到大数据同步项目**。
 2. 选择源和目标。

      1. 在 **选择源和目标** 页面，选择 **同步类型** 为 **OB → Kafka**。
      2. 设置 **同步链路名** **称**。

        **说明**

        同步链路名称支持中文、数字和英文字母的组合，且不得超过 64 个字符。
      3. 单击 **标签** 文本框，在下拉列表中选择目标标签。您也可以单击 **管理标签**，进行新建、修改和删除。详情请参见 [通过标签管理数据同步项目](https://www.oceanbase.com/docs/enterprise-oms-doc-cn-10000000000373406)。
      4. 设置 **源端**。

        | 参数 | 描述 |
        | --- | --- |
        | 表类型 | 包括 **单表** 和 **多表**。 |
        | 数据源 | 如果您已创建 OceanBase 数据源，请从下拉列表中进行选择。如果未创建，请单击下拉列表中的 **添加数据源**，在右侧对话框进行添加。参数详情请参见 [添加 OceanBase 数据源](https://www.oceanbase.com/docs/enterprise-oms-doc-cn-10000000000373440)。 |
        | 数据表 | 选择 **表类型** 为 **单表** 时，在 **数据表** 中输入关键字搜索源端数据表。 选择 **表类型** 为 **多表** 时，在 **数据表** 区域的左侧选中需要同步的数据表，单击 **>**，将其添加至右侧列表中。 **注意** 待同步的表名和其中的列名不能包含中文字符。 |
        | 分片列 | 在分片列下拉列表中选择目标分片列。您可以选择多个字段作为分片列，该参数为选填。 **注意** 仅选择表类型为单表时，会显示该参数。 请保证分片列的正确性，分片列填写错误会导致数据同步项目失败。 |
        | ETL 选项 | 同步 OceanBase 数据库的数据至 Kafka 支持简单的 ETL。 您可以单击 **ETL 选项**。在 **ETL 选项** 对话框中，您可以输入标准 SQL 语句的 `WHERE` 条件来配置行过滤。只有满足 `WHERE` 条件的数据才会被同步至目标数据源，以实现数据的行过滤。 **注意** 仅选择表类型为单表时，会显示该参数。 |

        同步 OceanBase 数据库多表数据至 Kafka 时，OMS 支持通过文本导入对象。操作如下：

             1. 在 **数据表** 区域的右侧列表中，单击 **导入对象**。
             2. 在对话框中，单击 **确定**。

               **注意**

               导入会覆盖之前的操作选择，请谨慎操作。
             3. 在 **导入迁移对象** 对话框中，输入需要迁移的对象。例如，`SCHEMA.TB1 | SCHEMA.TB2 |`。

               建议迁移对象不超过 1 万个。
             4. 单击 **检验合法性**。
             5. 通过合法性的检验后，单击 **确定**。
      5. 设置 **目标端**。

        | 参数 | 描述 |
        | --- | --- |
        | 数据源 | 如果您已创建 Kafka 数据源，请从下拉列表中进行选择。如果未创建，请单击下拉列表中的 **添加数据源**，在右侧对话框进行添加。参数详情请参见 [添加 Kafka 数据源](https://www.oceanbase.com/docs/enterprise-oms-doc-cn-10000000000373424)。 如果您需要查看 OMS 中所有已创建的数据源，请单击页面下方的 **查看数据源**。 |
        | Topic | Kafka 中的主题（Topic）。您可以输入一个已存在的 Topic，也可以在 Kafka Server 配置自动创建 Topic 属性，此处输入需要创建的 Topic。 请确保 Topic 名称符合 Kafka 规范，如果不符合命名规范，会导致数据同步任务失败。Kafka 的 Topic 命名规范请参见 Kafka 官方文档。 |
      6. 单击 **下一步**。
 3. 设置同步类型。

      1. 设置源端同步类型。

        | 参数 | 描述 |
        | --- | --- |
        | 同步类型和配置 | 目前仅支持 **增量同步**。增量同步的 DML 包括 **Insert**、**Delete** 和 **Update**，默认全部勾选。 |
        | 同步起始位点 | 指定同步某个时间节点之后的数据，默认为当前系统时间。您可以选择时间节点，也可以直接输入时间戳。 该位点与当前归档日志的保留时间密切相关。如果无特殊要求，可以从当前位点开始启动。 |
      2. 设置目标端同步类型。

        确认数据同步的目标端自建 Kafka 实例的相关参数，包括 **request.timeout.ms**、**max.block.ms**、**acks**、**max.in.flight.requests.per.connection**、**delivery.timeout.ms** 和 **序列化类型**。

        目标端同步类型均为默认值，且不支持自定义。该配置可以确保数据同步过程中，同步至 Kafka 的数据不丢失。

        目标端自建 Kafka 实例的参数说明如下。

        | 参数 | 描述 |
        | --- | --- |
        | request.timeout.ms | 表示消费者在重新发送请求或达到最大重试次数之前，等待相应请求的最大时间。默认值为 `IntMax`。 |
        | max.block.ms | 该参数指定了在调用 `send()` 方法或使用 `partitionsFor()` 方法获取元数据时生产者的阻塞时间。当生产者的发送缓冲区已满，或者没有可用的元数据时，这些方法就会阻塞。在阻塞时间达到 `max.block.ms` 时，生产者会抛出超时异常。默认值为 `IntMax`。 |
        | acks | 生产者在考虑完成请求之前要求 Leader 收到的确认的数量。该参数控制了发送记录的持久性。默认值为 `ALL`。 |
        | max.in.flight.requests.per.connection | 该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。默认值为 1。 |
        | delivery.timeout.ms | 发送消息上报成功或失败的最大时间。默认值为 `IntMax`。 |
        | 序列化类型 | 默认值为 `default`。 |
      3. 单击 **预检查**。
 4. 系统自动完成对数据同步项目的预检查。

   在 **预检查** 环节，OMS 会检测和目标端 Kafka 实例的连接情况。如果预检查报错，请排查并处理问题后，重新执行预检查，直至预检查成功。
 5. 单击 **启动任务**，启动该项目的增量同步等各项任务。

   如果您暂时无需启动任务，请单击 **保存**，跳转至数据同步项目的详情页面，您可以根据需要手动启动数据同步项目。如果您需要查看详情，请参见 [查看数据同步项目详情](https://www.oceanbase.com/docs/enterprise-oms-doc-cn-10000000000373403)。

 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线