---
title: "创建并保存 OceanBase 数据库至 Kafka 的数据同步项目 | OceanBase 文档中心"
description: 创建并保存 OceanBase 数据库至 Kafka 的数据同步项目 本文为您介绍 OMS 提供的 OpenAPI：创建并保存 OceanBase 数据库至 Kafka 的数据同步项目。 功能说明 本接口用于创建并保存 OceanBase 数据库至 Kafka 数据同步项目。 接口说明 接口约束 本接口适用于 OMS…
image: https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*OSPzQ6GUQF4AAAAAQHAAAAgAeiGDAQ/original
---
切换语言

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

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

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

更新时间：2023-06-08 10:49:34

本文为您介绍 OMS 提供的 OpenAPI：创建并保存 OceanBase 数据库至 Kafka 的数据同步项目。

## 功能说明

本接口用于创建并保存 OceanBase 数据库至 Kafka 数据同步项目。

## 接口说明

### 接口约束

本接口适用于 OMS V2.1.2 及以上版本。

### 请求路径

`POST /api/omsp/v2/sync/ob-kafka/projects`

### 请求参数

**Request Body**

```shell
{
  "name": "string",
  "sourceEndpointId": "string",
  "destEndpointId": "string",
  "kafkaTopic": "string",
  "recordTypes": [
    {}
  ],
  "databaseVOS": [
    {
      "databaseName": "string",
      "databaseId": "string",
      "mappingDatabaseName": "string",
      "tables": [
        {
          "tableName": "string",
          "tableId": "string",
          "mappingTableName": "string",
          "tableETL": {
            "tenantName": "string",
            "database": "string",
            "destDatabase": "string",
            "tableName": "string",
            "destName": "string",
            "whereClause": "string",
            "filterColumns]": [
              "string"
            ],
            "shardColumns": [
              "string"
            ],
            "logicTableId": "string",
            "sourceEndpointId": "string"
          }
        }
      ]
    }
  ],
  "startTimestamp": "integer[int64]"
}

```

| 参数 | 类型 | 必选 | 描述 |
| --- | --- | --- | --- |
| name | string | Y | 数据同步项目的名称。 |
| sourceEndpointId | string | Y | 源端数据源 ID，以"e_"开头。在创建 OceanBase 数据源的返回值中的 ID 字段获取。 |
| destEndpointId | string | Y | 目的端数据源 ID，以"e_"开头。在创建 Kafka 数据源的返回值中的 ID 字段获取。 |
| kafkaTopic | string | Y | Kafka Topic 名称。 |
| recordTypes | array | Y | 需要同步的增量 DML 记录类型。 支持 `INSERT`、`DELETE` 和 `UPDATE` 三种枚举值的组合。 |
| databaseVOS | array | Y | 需要同步的库表信息。具体包含的信息请参见下面的《DatabaseVO 数据结构》。 |
| startTimestamp | long | Y | 同步详细起始时间点，通常设置为当前时间。其设置不超过一天前的时间点，否则可能导致出现数据库的增量数据已经删除的问题。 **注意** 只有增量同步时，需要设置该参数。 |

**DatabaseVO 数据结构**

| 参数 | 类型 | 必选 | 描述 |
| --- | --- | --- | --- |
| databaseName | string | Y | 数据库的名称。 |
| databaseId | string | Y | 数据库的唯一标识。 |
| mappingDatabaseName | string | Y | 数据库的映射名称。 |
| tables | array | Y | 需要同步的表信息。具体包含的信息请参见下面的《TableVO 数据结构》。 |

**TableVO 数据结构**

| 参数 | 类型 | 必选 | 描述 |
| --- | --- | --- | --- |
| tableName | string | Y | 表名称。 |
| tableId | string | N | 逻辑表的唯一标识。 |
| mappingTableName | string | Y | 数据表的映射名称。 |
| tableETL | Object | Y | 逻辑表过滤条件 ETL。具体包含的信息请参见下面的《OmsSyncConditionVO 数据结构》。 |

**OmsSyncConditionVO 数据结构**

| 参数 | 类型 | 必选 | 描述 |
| --- | --- | --- | --- |
| tenantName | string | N | 租户的名称。 |
| database | string | N | 源端数据库的名称。 |
| destDatabase | string | N | 目标端数据库的名称。 |
| tableName | string | N | 源端的表名称。 |
| destName | string | N | 目标端的表称或 Topic 名称。 |
| whereClause | string | N | SQL 语法。 |
| filterColumns | array[string] | N | 过滤列。 |
| shardColumns | array[string] | N | 分片键列。 |
| logicTableId | string | N | 逻辑表的 Table ID。 |
| sourceEndpointId | string | N | 物理表的 EndPoint ID。 |

### 返回结果

返回结果中 `success` 为 `true` 代表创建成功，此时任务未开始执行，您需要另外执行启动接口。

您需要记录当前接口返回值中的 `data` 字段，代表当前创建的任务 ID。在后续的启动、暂停和查询任务状态等接口中，均需要使用该值。

| 参数 | 类型 | 描述 |
| --- | --- | --- |
| success | boolean | 是否调用成功。 |
| code | string | 状态码。 |
| message | string | 描述信息。 |
| advice | string | 建议。 |
| requestId | string | 请求 ID。 |
| pageNumber | integer[int32] | 当前页数。 |
| pageSize | integer[int32] | 每页条数。 |
| totalCount | integer[int64] | 总记录数。 |
| cost | string | 请求耗时。 |
| data | Object | 对于不同的接口，本字段包含的信息不同。 |

## 示例

### 请求示例

```shell
curl --location --request POST 'http://xxx.xxx.xxx.1:8089/api/omsp/v2/sync/ob-kafka/projects' \
--header 'Authorization: Basic YWRt=' \
--header 'Content-Type: application/json' \
--data '{
  "name": "name",
  "sourceEndpointId": "sourceEndpointId",
  "destEndpointId": "destEndpointId",
  "shardColumns": [
    "TENANT_NAME"
  ],
  "kafkaTopic": "test",
  "recordTypes": [
    "INSERT",
    "DELETE",
    "UPDATE"
  ],
  "databaseVOS": [
    {
      "databaseName": "databaseName",
      "tables": [
        {
          "tableName": "tableName",
          "tableETL": {}
        }
      ]
    }
  ],
  "startTimestamp": 1637568354
}'

```

### 返回示例

```json
{
  "success": true,
  "code": null,
  "message": null,
  "advice": null,
  "requestId": null,
  "pageNumber": null,
  "pageSize": null,
  "totalCount": null,
  "cost": null,
  "data": "sync_38t**"
}

```

 上一篇 下一篇 ![有帮助](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) 咨询热线