# 使用CDL从PgSQL同步数据到Kafka
#### 操作场景
本章节指导用户通过MRS 3.2.0版本开启Kerberos认证的集群的CDLService WebUI界面，从PgSQL导入数据到Kafka。
#### 前提条件
- 集群已安装CDL、Kafka服务且运行正常。
- PostgreSQL数据库需要修改预写日志的策略，操作步骤请参考[PostgreSQL数据库修改预写日志的策略](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24124.html#mrs_01_24124__li1868193914169)。
- 在FusionInsight Manager中创建一个人机用户，例如"cdluser"，加入用户组**cdladmin、hadoop、kafka**，主组选择"cdladmin"组，关联角色"System_administrator"。
 
#### 操作步骤
1. 使用**cdluser**用户登录FusionInsight Manager（首次登录需要修改密码），选择"集群 \> 服务 \> CDL"，在CDL"概览"界面单击"CDLService UI"右侧的超链接，进入CDL原生界面。
2. 选择"连接管理 \> 新增连接"，进入"新增连接"参数配置窗口，参考下表，分别新增"pgsql"和"kafka"连接，相关数据连接参数介绍请参见[创建CDL数据库连接](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24238.html)。
   
   表1PgSQL数据连接配置参数 
   | 参数名称        | 示例            |
   |:---|:---|
   | Link Type   | pgsql         |
   | Name        | pgsqllink     |
   | Host        | 10.10.10.10   |
   | Port        | 5432          |
   | DB Name     | testDB        |
   | User        | user          |
   | Password    | *user* *用户密码* |
   | Description | -             |
      
   表2Kafka数据连接配置参数 
   | 参数名称        | 示例        |
   |:---|:---|
   | Link Type   | kafka     |
   | Name        | kafkalink |
   | Description | -         |
      
   
   
3. 参数配置完成后，单击"测试连接"，检查数据连通是否正常。 
   连接校验通过后，单击"确定"完成数据连接创建。
   
   
4. 在"作业管理"页面单击"新建作业"。在"新建作业"窗口中填写配置。单击"下一步"，进入作业参数配置页面。
   
   其中：
   
   | 参数名称 | 示例               |
   |:---|:---|
   | Name | job_pgsqltokafka |
   | Desc | xxx              |
      
   
   
5. 配置PgSQL作业参数。 
   1. 在作业参数配置页面，选取左侧"pgsql"图标拖入右侧编辑区域，然后双击此图标进入PgSQL作业参数配置窗口，相关作业参数介绍请参见[创建CDL数据同步任务作业](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24239.html)。
      表3PgSQL作业参数 
      | 参数名称                  | 示例                   |
      |:---|:---|
      | Link                  | pgsqllink            |
      | Tasks Max             | 1                    |
      | Mode                  | insert、update、delete |
      | Schema                | public               |
      | dbName Alias          | cdc                  |
      | Slot Name             | test_slot            |
      | Slot Drop             | 否                    |
      | Connect With Hudi     | 否                    |
      | Use Exist Publication | 是                    |
      | Publication Name      | test                 |
         
      
   
   2. 单击"+"按钮展开更多选项。 ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/zh-cn_image_0000001724747081.png)
      ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
      - "WhiteList"：输入数据库中的表（如myclass）
      
      - "Topic Table Mapping"：第一个框输入topic名（与[4]中作业名称"Name"的值不能一样，例如myclass_topic）。 第二个框输入表名（例如myclass。该值与第一个框的topic只能是一对一的关系）。
        
   
   3. 单击"确定"，PgSQL作业参数配置完成。
   
   
   
   
6. 配置Kafka作业参数。 
   1. 在作业参数配置页面，选取左侧"kafka"图标拖入右侧编辑区域，然后双击此图标进入Kafka作业参数配置窗口。参考[表4]进行参数配置。
       表4Kafka作业参数 
      | 参数名称 | 示例        |
      |:---|:---|
      | Link | kafkalink |
         
      
   
   2. 单击"确定"，完成Kafka作业参数配置。
   
   
   
   
7. 作业参数配置完成后，拖拽图标将作业进行关联，然后单击"保存"，作业配置完成。 
   ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/zh-cn_image_0000001676747874.png)
   
   
8. 在"作业管理"的作业列表中，找到创建的作业名称，单击操作列的"启动"，等待作业启动。 
   观察数据传输是否生效，例如在PgSQL数据库中对表进行插入数据操作，然后参考[查看Kafka数据生产消费详情](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24138.html)进入KafkaUI界面查看Kafka的Topic中是否有数据生成。