GaussDB集中式/分布式同步到Doris作业配置

GaussDB即GaussDB集中式/分布式数据库，本章节以GaussDB为例。

数据库账号权限要求

在使用Migration进行同步时，源端和目的端所使用的数据库账号需要满足以下权限要求，才能启动实时同步任务。不同类型的同步任务，需要的账号权限也不同，详细可参考下表进行赋权。

表2 数据库账号权限

类型名称	权限要求
源数据库连接账号	需要REPLICATION权限或继承了内置角色gs_role_replication的权限、DATABASE的CONNECT权限、SCHEMA的USAGE权限、表的SELECT权限、序列的SELECT权限，同时需要有查询pg_ls_waldir()函数的权限，具体操作请参见如何增加PostgreSQL、Gaussdb数据源额外权限？。 需要具备远程连接权限。 远程连接权限的添加方法： 集中式：在源数据库的“gs_hba.conf”配置文件添加配置“host all <user> 0.0.0.0/0 sha256”和“host replication <user> 0.0.0.0/0 sha256”。 分布式：在源数据库CN节点的“gs_hba.conf”配置文件添加配置“host all <user> 0.0.0.0/0 sha256”；在源数据库DN节点的“gs_hba.conf”配置文件添加配置“host all <user> 0.0.0.0/0 sha256”和“host replication <user> 0.0.0.0/0 sha256”。 在源库使用系统管理员用户执行语句“select pg_reload_conf();”生效，或重启数据库实例生效。 需保证安全组和防火墙策略放开对应连接端口。 安全组和防火墙开放策略：数据库连接已放开普通连接和复制连接端口。 集中式：复制连接端口为普通连接端口+1。 分布式：复制连接端口为CN/DN连接端口+1，在源数据库执行语句“select * from pgxc_node”可查看DN连接端口。
目标数据库连接账号	目标数据库的每张表必须具有如下权限：INSERT、SELECT、UPDATE、DELETE、CONNECT、CREATE。

• 建议创建单独用于Migration任务连接的数据库账号，避免因为数据库账号密码修改，导致的任务连接失败。
• 连接源和目标数据库的账号密码修改后，请同步修改管理中心对应的连接信息，避免任务连接失败后自动重试，导致数据库账号被锁定影响使用。

1. 参见新建实时集成作业创建一个实时集成作业并进入作业配置界面。
2. 选择数据连接类型：源端选PostgreSQL，目的端选Doris。
   图1 选择数据连接类型
   

3. 选择集成作业类型：同步类型默认为实时，同步场景包含整库场景。
   图2 选择集成作业类型
   
   

   同步场景相关介绍请参见同步场景。

4. 配置网络资源：选择已创建的GaussDB（RDS PostgreSQL）、Doris数据连接和已配置好网络连接的migration资源组。
   图3 选择数据连接及migration资源组
   
   

   无可选数据连接时，可单击“新建”跳转至管理中心数据连接界面，单击“创建数据连接”创建数据连接，详情请参见配置DataArts Studio数据连接参数进行配置。

   无可选migration资源组时，可单击“新建”跳转至购买migration资源组页面创建migration资源组配置，详情请参见购买创建数据集成资源组增量包进行配置。

5. 检测网络连通性：数据连接和migration资源组配置完成后需要测试整个迁移任务的网络连通性，可通过以下方式进行数据源和migration资源组之间的连通性测试。
   • 单击展开“源端配置”触发连通性测试，会对整个迁移任务的连通性做校验。
   • 单击源端和目的端数据源和migration资源组中的“测试”按钮进行检测。
     

     网络连通性检测异常可先参考数据源和资源组网络不通如何排查？章节进行排查。

6. 配置源端参数。
   选择需要迁移的GaussDB库表。

   图4 选择库表
   

   库与表均支持自定义选择，即可选择一库一表，也可选择多库多表。

7. 配置目的端参数。
   • 源库表和目标匹配策略。
     • 库匹配策略。
       • 与来源库同名：数据将同步至与来源GaussDB库名相同的Doris库中。
       • 自定义：数据将同步至自行指定的Doris库中。
     • 表匹配策略。
       • 与来源表同名：数据将同步至与来源GaussDB表名相同的Doris表中。
       • 自定义：数据将同步至自行指定的Doris表中。
         图5 整库场景下源库表和目标匹配策略
         
         

         自定义匹配策略时，支持用内置变量#{source_db_name}和#{source_table_name}标志来源的库名和表名，其中表匹配策略必须包含#{source_table_name}。

   • Doris参数配置。

     高级配置：支持通过参数配置部分高级功能，参数详情可参考下表。

     表5 Doris高级配置一览表

     参数名	参数类型	默认值	单位	配置说明
     doris.request.connect.timeout.ms	int	30000	ms	Doris连接超时时间。
     doris.request.read.timeout.ms	int	30000	ms	Doris读取超时时间。
     doris.request.retries	int	3	-	Doris请求失败重试次数。
     sink.max-retries	int	3	-	数据库写入失败时的最大重试次数。
     sink.batch.interval	string	1s	h/min/s	异步线程写入数据的时间间隔。
     sink.enable-delete	boolean	true	-	是否启用删除功能，如果关闭，源端删除数据，目的端不会删除。
     sink.batch.size	int	20000	-	单次写（插入、更新、删除）数据的最大行数。
     sink.batch.bytes	long	10485760	bytes	单次写（插入、更新、删除）数据的最大字节数。
     logical.delete.enabled	boolean	false	-	逻辑删除开关，开启本开关后，需要目的端包含删除标记列，在源端删除数据，不会将目的端对应的数据删除，而是将删除标记列设为true，表示该数据在源端已不存在。
     logical.delete.column	string	logical_is_deleted	-	逻辑删除标记列名称，默认为logical_is_deleted，支持用户自定义。
     sink.keyby.mode	string	pk	-	doris多并发写时的分区方式，默认为pk(主键)，源端为kafka类型没有主键时，应选择table，以库表名进行分区。
     doris.sink.flush.tasks	int	1	-	单个taskmanager的flush并发数。
     sink.properties.format	string	json	-	Stream Load 使用的数据格式，可选择json/csv。
     sink.properties.Content-Encoding	string	-	-	HTTP头部消息体压缩格式，目前只支持 CSV 文件的压缩，支持gzip。
     sink.properties.compress_type	string	-	-	文件的压缩格式，目前只支持 CSV 文件的压缩。支持 gz, lzo, bz2, lz4, lzop, deflate 压缩格式。

8. 刷新源表和目标表映射，检查映射关系是否正确，同时可根据需求修改表属性、添加附加字段，并通过“自动建表”能力在目的端Hudi数据库中建出相应的表。
   图6 源表与目标表映射
   
   • 附加字段编辑：单击操作列“附加字段编辑”可为目的端的Doris表中增加自定义字段，同时附加字段也会额外加入到Doris表的建表中。用户可以在已有的源表字段基础上添加多个附加字段，并自定义字段名、选择字段类型、填写字段值。
     • 字段名称：目的端Doris表新增字段的名称。
     • 字段类型：目的端Doris表新增字段的类型。
     • （可选）字段类型长度：目的端Doris表新增字段类型的长度。
     • 字段值：目的端Doris表新增字段的取值来源。

       表6 附加字段取值方式

       类型	示例
       常量	支持数字、字母、中文、特殊字符。彩色表情字符可能导致作业提交失败，需谨慎使用。

   • 自动建表：单击“自动建表”可按照已配置映射规则在目的端数据库自动建表，成功后表建立方式会显示为使用已有表。
     

     • Migration仅支持自动建表，不支持自动建库和模式，需用户自行在目的端手动建出库和模式后再使用本功能建表。
     • 自动建表时对应的字段类型映射关系请参见字段映射关系章节。

9. 配置任务属性。

   表7 任务配置参数说明

   参数	说明	默认值
   执行内存	作业执行分配内存，跟随处理器核数变化而自动变化。	8GB
   处理器核数	范围：2-32。 每增加1处理核数，则自动增加4G执行内存和1并发数。	2
   并发数	作业执行支持并发数。该参数无需配置，跟随处理器核数变化而自动变化。	1
   自动重试	作业失败时是否开启自动重试。	否
   最大重试次数	“自动重试”为是时显示该参数。	1
   重试间隔时间	“自动重试”为是时显示该参数。	120秒
   是否写入脏数据	选择是否记录脏数据，默认不记录脏数据，当脏数据过多时，会影响同步任务的整体同步速度。 否：默认为否，不记录脏数据。 表示不允许脏数据存在。如果同步过程中产生脏数据，任务将失败退出。 是：允许脏数据，即任务产生脏数据时不影响任务执行。 允许脏数据并设置其阈值时： 若产生的脏数据在阈值范围内，同步任务将忽略脏数据（即不会写入目标端），并正常执行。 若产生的脏数据超出阈值范围，同步任务将失败退出。 说明： 脏数据认定标准：脏数据是对业务没有意义，格式非法或者同步过程中出现问题的数据；单条数据写入目标数据源过程中发生了异常，则此条数据为脏数据。 因此只要是写入失败的数据均被归类于脏数据。 例如，源端是VARCHAR类型的数据写到INT类型的目标列中，则会因为转换不合理导致脏数据不会成功写入目的端。用户可以在同步任务配置时，配置同步过程中是否写入脏数据，配置脏数据条数（单个分片的最大错误记录数）保证任务运行，即当脏数据超过指定条数时，任务失败退出。	否
   脏数据策略	“是否写入脏数据”为是时显示该参数，当前支持以下策略： 不归档：不对脏数据进行存储，仅记录到任务日志中。 归档到OBS：将脏数据存储到OBS中，并打印到任务日志中。	不归档
   脏数据写入连接	“脏数据策略”选择归档到OBS时显示该参数。 脏数据要写入的连接，目前只支持写入到OBS连接。	-
   脏数据目录	脏数据写入的OBS目录。	-
   脏数据阈值	是否写入脏数据为是时显示该参数。 用户根据实际设置脏数据阈值。 说明： 脏数据阈值仅针对每个并发生效。比如阈值为100，并发为3，则该作业可容忍的脏数据条数最多为300。 输入-1表示不限制脏数据条数。	100
   添加自定义属性	支持通过自定义属性修改部分作业参数及开启部分高级功能，详情可参见任务性能调优章节。	-

10. 提交并运行任务。

    作业配置完毕后，单击作业开发页面左上角“提交”，完成作业提交。

    图7 提交作业
    

    提交成功后，单击作业开发页面“启动”按钮，在弹出的启动配置对话框按照实际情况配置同步位点参数，单击“确定”启动作业。

    图8 启动配置
    

    表8 启动配置参数

    参数	说明
    同步模式	增量同步：从指定时间位点开始同步增量数据。 全量+增量：先同步全量数据，随后实时同步增量数据。

11. 监控作业。

    通过单击作业开发页面导航栏的“前往监控”按钮，可前往作业监控页面查看运行情况、监控日志等信息，并配置对应的告警规则，详情请参见实时集成任务运维。

    图9 前往监控