# 通用模板作业

通用模版作业不同于其他作业的可视化拖拽开发方式，而是采用界面向导配置方式进行操作。随着产品的发展，通用模版类型会不断扩展。

目前，DWS 支持的通用模板有五种：

数据库同步
CDC增量同步
数据库到Hive同步
基于时间戳的数据库增量同步
基于时间戳的数据库增量同步到Hive

template

# 数据库同步

数据库同步可以实现单表、多表、整库的迁移，将全部数据一次性从源表迁移到目标表。

单表：只选择一张表名，可快速实现两个表之间的数据批量迁移。
多表：选择多张表名，可快速实现多个表之间的数据批量迁移。
整库：全选表名。整库同步是帮助用户提升数据同步效率的工具，通过整库同步可快速实现两个库之间的数据批量迁移，节省任务配置工作，快速提升数据迁移效率。

数据库同步一共有三步：

第一步：选择来源库、表名称。
第二步：选择目标库、表名称。
第三步：来源库表名称与目标库表名称的映射。

multi

# CDC增量同步

增量同步与多表迁移的向导界面类似，且都支持多表增量同步。与多表迁移的不同点在于，增量同步不是一次性将数据从源表同步到目标表，而是源表的数据发生变化（增加、修改、删除）时才会同步到目标表。

⚠️ 提示：

1、目标表不存在时，模型运行时会自动创建表，和源表结构一致；

2、目标表存在时，要确保源表与目标表的数据结构、字段类型一致，且定义主键。同时源表开启了 cdc 配置。

增量同步支持的数据源如下：

数据源类型	版本	驱动
MySQL-CDC	MySQL: 5.6,5.7,8.0.x	JDBC Driver: 8.0.28
PostgreSQL-CDC	PostgresSQL: 9.6, 10, 11, 12, 13,14	JDBC Driver: 42.5.1
SQLServer-CDC	Sqlserver: 2012, 2014, 2016, 2017, 2019	JDBC Driver: 9.4.1.jre8
MongoDB-CDC	MongoDB: 3.12.11	JDBC Driver: 3.12.11

CDC 增量同步作业需要提前对数据库进行配置，具体的配置方法如下：

# MySQL-CDC 配置

1、修改 MySQL 配置，开启 binlog 功能，并设置 binlog 格式为 row 模式。

binlog 是 MySQL 的二进制日志文件，它记录了数据库中的所有变更操作，Mysql CDC Connector 会通过 binlog 来捕获数据的变化。

参考配置如下：

[mysqld]
 
# ----------------------------------------------
# Enable the binlog for replication & CDC
# ----------------------------------------------
 
# Enable binary replication log and set the prefix, expiration, and log format.
# The prefix is arbitrary, expiration can be short for integration tests but would
# be longer on a production system. Row-level info is required for ingest to work.
# Server ID is required, but this will vary on production systems
server-id         = 223344
log_bin           = mysql-bin
expire_logs_days  = 1
binlog_format     = row
 
# enable gtid mode
gtid_mode = on
enforce_gtid_consistency = on

2、添加权限（如果是非 root 用户）

MySQL Binlog 权限需要三个权限 SELECT, REPLICATION SLAVE, REPLICATION CLIENT：

Select 权限代表允许从表中查看数据。

Replication client 权限代表允许执行show master status,show slave status,show binary logs 命令。

Replication slave 权限代表允许slave主机通过此用户连接master以便建立主从复制关系。

3、执行命令（如果是非 root 用户）

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'dws'@'%' IDENTIFIED BY 'dws';

# PostgreSQL-CDC 配置

1、修改 postgresql.conf 配置文件，参考配置如下：

# 更改wal日志方式为logical（方式有：minimal、replica 、logical ）
wal_level = logical

# 更改solts最大数量（默认值为10），flink-cdc默认一张表占用一个slots 
max_replication_slots = 20
 
# 更改wal发送最大进程数（默认值为10），这个值和上面的solts设置一样
max_wal_senders = 20

# 中断那些停止活动超过指定毫秒数的复制连接，可以适当设置大一点（默认60s，0表示禁用） 
wal_sender_timeout = 180s

2、开启权限

-- 更改复制标识包含更新和删除之前值
ALTER TABLE table_name  REPLICA IDENTITY FULL;
-- 查看复制标识（为f标识说明设置成功）
select relreplident from pg_class where relname='table_name';

# SQL Server-CDC 配置

1、开启SQL Server CDC的必要条件

需要开启数据库 CDC
需要开启表 CDC
需要开启代理服务
磁盘要有足够的空间，保存日志文件
表必须要有主键或者是唯一索引

2、参考文档：https://support.huaweicloud.com/intl/zh-cn/usermanual-roma/fdi-ug-202009081.html

# MongoDB-CDC 配置

1、开启Mongodb CDC的必要条件

MongoDB version: MongoDB version >= 4.0.
Cluster deployment: replica sets or sharded clusters
Storage Engine: WiredTiger Storage Engine.
Permissions:changeStream and read（执行如下语句）

use admin;
db.createRole(
    {
        role: "strole",
        privileges: [{
            resource: { db: "", collection: "" },
            actions: [
                "splitVector",
                "listDatabases",
                "listCollections",
                "collStats",
                "find",
                "changeStream" ]
        }],
        roles: [
            { role: 'read', db: 'config' }
        ]
    }
);
db.createUser(
  {
      user: 'stuser',
      pwd: 'stpw',
      roles: [
         { role: 'strole', db: 'admin' }
      ]
  }
);

2、配置 MongoDB 副本集创建 mongo1.conf，参考配置如下：

# 端口默认 
port = 27017
#数据目录
dbpath = /usr/local/Cellar/mongodb-community/7.0.0/data/db
#日志所在目录
logpath = /usr/local/Cellar/mongodb-community/7.0.0/data/logs/mongodb.log
#日志输出方式
logappend = true
# 在后台启动
fork=true
# 0.0.0.0 表示任意IP均可连接
bind_ip=0.0.0.0
#副本集名称
replSet=rs0

3、启动 MongoDB 节点，参考命令如下：

cd /usr/local/Cellar/mongodb-community/7.0.0/bin
启动命令：
mongod -f mongo.conf

4、初始化副本集

# 连接mongodb
./mongo mongodb://127.0.0.1:27017
use admin
# 设置变量
rs0= {
   _id: "rs0",
   members: [
     {
      _id: 0,
      host: "127.0.0.1:27017"
     },
    ]
 }
# 执行初始化
rs.initiate(rs0)