实例管理

实例管理概述

数据迁移功能采用独立集群的方式为用户提供安全可靠的数据迁移服务，各集群之间相互隔离，不可互相访问。其中实例管理通过购买GDS-Kafka实例帮助用户创建、管理集群。GDS-Kafka的工作方式是从Kafka中消耗数据并缓存，当达到设置好的时间或数据量之后，通过COPY写入GaussDB(DWS)临时表，再从临时表进行插入或更新操作。

• Kafka的消息生产端必须按照一定的格式要求进行数据生产，其中消息格式由“kafka.source.event.type”配置参数指定。当前支持的消息格式详见GDS-Kafka支持的消息格式。
• GDS-Kafka支持直接insert（仅限无主键表）和merge覆盖更新两种入库模式，您可以根据DWS目标表的类型进行灵活配置，直接insert模式由于不涉及更新在性能上要更优一些。其中入库模式由“app.insert.directly”配置参数和有无主键共同决定，详见GDS-Kafka入库模式。

• GDS-kafka只支持目标表表名和字段全小写。
• GDS-Kafka的删除是根据扩展字段中的pos进行历史删除，如果入库数据中有delete操作，则必须使用扩展字段。

购买GDS-Kafka实例

使用数据迁移功能，首先需要购买一个GDS-kafka实例（集群）。集群实例可以为用户提供安全可靠的数据迁移服务，各集群之间相互隔离。

约束与限制

• 目前仅支持单机集群。
• 目前仅支持按需计费模式。

操作步骤

1. 登录GaussDB(DWS)控制台。
2. 在左侧导航栏选择“数据 > 数据集成 > 实例管理”，进入实例管理页面。
3. 单击页面右上角“购买GDS-Kafka实例”，进入购买页面配置集群参数。

   表1 参数说明

   参数名	参数解释	样例值
   CPU架构	CPU架构有： X86 鲲鹏 说明： X86和鲲鹏只是底层架构不一致，应用层不感知，sql语法一致，如果创建集群时提示X86资源不足，可以选择鲲鹏架构。	x86
   规格	请根据业务需求合理选择节点规格。	-
   存储	单个节点可用存储容量。	-
   当前规格	显示集群当前规格。	-
   名称	设置数据仓库集群的名称。 集群名称长度为4到64个字符，必须以字母开头，可以包含字母、数字、中划线或者下划线，不能包含其他的特殊字符。字母不区分大小写。	-
   版本	显示集群中安装的数据库实例版本。	-
   虚拟私有云	指定集群节点使用的虚拟专用网络，实现不同业务的网络隔离。 首次创建数据仓库集群时，如果未配置过虚拟私有云，可以单击“查看虚拟私有云”进入虚拟私有云管理控制台，新创建一个满足需求的虚拟私有云。	-
   子网	指定虚拟私有云的一个子网。 集群使用子网实现与其他网络的隔离，并独享所有网络资源，以提高网络安全。	-
   安全组	指定虚拟私有云的安全组。 安全组限制安全访问规则，加强集群与其它服务间的安全访问。	-
   公网访问	指定用户是否可以在互联网上使用客户端连接集群数据库。支持如下方式： “暂不使用”：暂不使用弹性IP。如DWS使用于生产环境，则需绑定ELB，则不选择使用弹性IP，通过绑定ELB后，再从ELB页面进行弹性IP绑定。 “现在购买”：用户指定弹性IP的带宽，系统将自动为集群分配独享带宽的弹性IP，通过弹性IP可以从互联网对集群进行访问。自动分配的弹性IP的带宽名称都是以集群名称开头的。 “使用已有”：为集群绑定指定的弹性IP。如果下拉框中没有可用的弹性IP，可以单击“创建弹性IP”进入弹性公网IP页面创建一个满足需要的弹性IP。带宽可根据用户需要设置。	-
   企业项目	配置集群所属的企业项目。已开通企业项目管理服务的用户才可以配置该参数。默认值为default。	default

4. 确认无误后，单击“立即购买”进行创建。

查看实例详情

在实例详情页面用户可以查看集群的详细信息，用户可以在此查看集群的基本信息、网络信息等。

操作步骤

1. 登录GaussDB(DWS)控制台。
2. 在左侧导航栏选择“数据 > 数据集成 > 实例管理”，进入实例管理页面。
3. 单击指定实例名称，进入实例详情页面。
   图1 查看实例详情
   

GDS-Kafka支持的消息格式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Record record = new Record();
// 设置目标表schema和table名称
record.setTableName("public.person");
// 设置字段列表
List<Field> fields = new ArrayList<>();
fields.add(new Field("id", 0));
fields.add(new Field("name", 1));
fields.add(new Field("age", 2));
record.setFields(fields);
// 设置表记录更新前的字段值列表
List<Object> before = new ArrayList<>();
before.add(new Integer(0, "0"));
before.add(new Character("utf-8", ByteBuffer.wrap("a".getBytes(StandardCharsets.UTF_8))));
before.add(new Integer(0, "18"));
record.setBeforeImages(before);
// 设置表记录更新后的字段值列表
List<Object> after = new ArrayList<>();
after.add(new Integer(0, "0"));
after.add(new Character("utf-8", ByteBuffer.wrap("b".getBytes(StandardCharsets.UTF_8))));
after.add(new Integer(0, "20"));
record.setAfterImages(after);
// 设置操作类型
record.setOperation("U");
// 设置操作时间
record.setUpdateTimestamp(325943905);
// 将Record对象序列化为byte[]数组
byte[] msg = Record.getEncoder().encode(record).array();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
"table": "public.person",
"op_type": "U",
"op_ts": "1668426344",
"current_ts": "1668426344",
"before": {
"id":"1",
"name":"a",
"age": 18
},
"after": {
"id":"1",
"name":"b",
"age": 20
}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

{
"header": {
"thing_id":"a0001",
"instance_id":"1",
"thing_model_name":"computer",
"timestamp":"1668426344"
},
"body": {
"status":"Normal",
"temperature":"10",
"working_time":"10000"
},
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

{
"header": {
"thing_id":"a0001",
"instance_id":"1",
"thing_model_name":"computer",
"timestamp":"1668426344"
},
"body": {
"status":"Normal",
"temperature":"10",
"property":{
  "key1":"1",
  "key2":2
},
"working_time":"10000"
},
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
"event_id":"1",
"event_name":"test",
"start_time":"1970-1-1T00:00:00.000Z",
"end_time":"1970-1-1T00:00:00.000Z",
"fields":{
    "field1":"value1",
    "field2":2
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
"event_id":"1",
"event_name":"test",
"start_time":"1970-1-1T00:00:00.000Z",
"end_time":"1970-1-1T00:00:00.000Z",
"fields":{
    "field1":"value1",
    "field2":2,
    "field3":{
       "key1":"1",
       "key2":2
       }
    }
}

GDS-Kafka入库模式

GDS-Kafka的数据入库都是先将数据copy至临时表，然后再根据客户的使用场景以及目标表有无主键进行merge或者insert，详见下表：