一、介绍

 ZCBUS平台中，ZDT【ZCBUS DATA TRANSLATE】模块，为运行在软件内部模块，为数据流提供数据过滤，转换，整理，事件触发等多种动作提供数据处理服务的组件，完全在KAFKA内部完成。
ZDT模块自动衍生与ETL组件模块，所以在系统中，ZDT组件也成为ETL组件。
数据处理加工说明：

二、ZDT配置

 参考ZCBUS平台中，增加数据订阅服务管理，当选择客户端类型的时候，选择为数据加工客户端，即使用ZDT数据模块。

 每个容器内可以含有多个数据加工客户端，可以实现TOPIC的一级、N级数据交换。

三、ZDT操作介绍

1、加工管理

• 加工管理：添加、删除、修改加工客户端
• 数据库管理：在进行数据加工时可以选择是否通过数据库做数据处理操作。非数据库，则全部为基础条件设置，实现数据过滤，转换等规则。如果是数据库，则使流数据可以通过某个SQL与指定数据库关联，实现数据转换规则，当与数据库关联之后，则数据交换服务将会根据数据库的关系实现更为灵活的数据处理处理机制。
• 表属性自动更新：
• 统计：查看订阅通道当中表加工的状态

2、通道配置说明

 将赋予数据加工客户端-数据应用中的策略和主控加工通道相关联，合称为准备转换的列表。再在数数据表中，对每个表执行ETL相关配置。

• 配置：将策略与加工通道关联，再在数据表中，对每个表执行ETL相关规则配置
• 新增：当某一通道内处理的表过多时可能会造成数据延时过大，可在该策略下新增多个通道，用以分担压力，新增的通道共用同一通道属性参数
• 解绑：将策略与加工通道解除关联，如需做此操作，请确认加工表的状态已重置，且通道处于停止状态

3、通道操作说明

• 属性：数据库和数据应用策略之间桥梁连接作用
  • 1. 数据应用过程中，多线程策略管理服务
  • 2. 数据应用过程中，针对整个策略中的数据，在通道层，可以整体配置表附加信息
       • 加载过程中，DDL的控制粒度，是否执行DDL等操作
       • 加载过程中，是否增加辅助列：合并分区列数据、源端生产操作提交时间、数据最后操作类型、目标端数据加载时间、数据首次插入到目标端需要的时间字段。
  • 3. 数据加载的过程中，是否完全过滤DDl操作等
  • 4. 控制全量、增量数据加载方式
• 数据表：对需要加工的数据表进行加工规则配置及开启数据加工等操作
  • 操作类型：操作数据表开启全量订阅及恢复、暂停的状态
  • 配置类型：更改数据表的表名大小写及表名、库名等信息
  • ETL配置：可对数据表添加数据加工的规则，如数据过滤，数据处理等
  • ETL配置SQL关联：：
  • 通道转移：当同一策略内存在多个加工通道时，可使用此功能将数据表表转移到其他加工通道中
  • 异常信息批处理：针对通道可出现多个加工表异常时，可对异常的表进行批量处理
• 监控：监控加工数据表的全量及增量加工状态
• 平衡：当策略内存在多个加工通道时，可根据加工表的繁忙、延时等信息平衡加工通道内的表数量

四、转换规则介绍

数据过滤：

• 数据分流操作类型支持【INSERT/UPDATE/DELETE/ALL】

   PASS(SQLTYPE=(ON=INSERT,IGNORE=UPDATE,ON=UPDATE),CONDITION=( id > 10 ))

• 此用例针对操作过程中，INSERT/DELETE按照条件过滤，UPDATE全部忽略，不进行传输 ON=? 代表支持那个那种类型的操作进行配置。INSERT 插入 UPDATE 更新 DELETE 删除 ALL所有操作

• 过滤条件支持WHERE条件组合，以及SQL语句整合
  • 条件过滤

    PASS(SQLTYPE=(ON=INSERT,ON=UPDATE,ON=DELETE),CONDITION=(flag=1 and area='Beijing'))

    SQL语句整合

    PASS(SQLTYPE=(ON=INSERT,ON=UPDATE,ON=DELETE),SQLCONDITION=(SELECT 1 FROM DEMO.TEST001 WHERE FLAG=#FLAG# AND AREA=#AREA#))

    注意：
    #ZCBUS_SCHEMA# 代表原数据用户名
    #ZCBUS_TABLENAME# 代表原数据表名

数据处理：

• 复制数据转换，脱敏规则
  • 以列为单位进行配置，可以添加、删除、修改、重命名列
  • COLUMN_NAME为需要设置到列名
  • OPTYPE为ADD/RENAME/MODIFY/DROP操作
  • DATA_TYPE=STRING/NUMBER/DATE类型
  • FUNCTION分为FUNCTION/SQLFUNCTION两种
    • FUNCTION 为函数转换，相对比较固定，简单，易操作；支持函数嵌套组合
      • SQLFUNCTION为SQL语句函数转换，可以依赖配置表，多个MAP信息转换等
        注：针对SQLFUNCTION的转换，可以持久化内存，亦可以每条数据均通过SQL进行转换
• 函数列数据转换

  COLMAP(column_name=ob_fundcode_0235,optype=modify,data_type=string,function=@strcat(@COMPUTEFLOAT(id*3.343),'-',@REPLACEALL(ob_object_id,'1','2))

• 通过SQL语句进行数据转换【支持MYSQL/ORACLE/SQLSERVER】

  COLMAP(column_name=ob_fundcode_0235,optype=modify,data_type=string,sqlfunction=select abc ob_fundcode_0235 from ob_fund_0032 where id = #id#)

• 修改列名

  COLMAP(column_name=c10,optype=rename,data_type=string,after_column_name=col6)

• 注：查询出来的列，必须和COLUMN_NAME列匹配，否则会匹配失败

• 附加：可以支持列和其他字段组合，支持数据库相关的所有的函数调用

  COLMAP(column_name=ob_fundcode_0235,optype=modify,data_type=string,sqlfunction=select abc+#codeid# ob_fundcode_0235 from ob_fund_0032 where id = #id#)

• Optype：包括add/modify/drop/rename四种对列操作的类型

事件触发

• 根据传输的数据，支持触发类型
  SQL语句、
  存储过程、
  SHELL、
  BAT等
• 触发SQL语句【支持MYSQL/ORACLE/SQLSERVER】

  COLTRIGGER(type=sql,text=insert into f003 select * from demo.test004 where id = #ob_fundcode_test_001# and objectid=#ob_object_id# and c2 = #c2# and type=‘001’)

• 触发调用存储过程

  COLTRIGGER(type=procedure,text=proc_test(#c2#,#ob_fundcode_test_001#,#ob_object_id#))

• SHELL触发调用

  COLTRIGGER(type=shell,text=sh /zbomc/abc/ccc.sh #ob_fundcode_test_001# #ob_object_id#)

• BAT出发调用

  COLTRIGGER(type=cmd,text=d:\\adjuct.bat #codeid#)

• 注：上述触发条件中，#字段# 为输入的条件，可以根据实际情况进行参数传递

SQL规则生成TOPIC

根据设定的SQL语句，复制表为主表，形成复制规则，每条记录根据SQL语句查询结果集，将结果集再次输入到指定TOPIC中。
在多表关联的过程中，目前主要是个别表，简单操作为主。

注意：
（1）SQLFUNCTION规则中，sql语句查询结果集中，返回列必须设置别名，作为新的数据流中的列名
（2）需要指定数据库/用户/schemas名称（例如：ds.t1 a, ds.t2 b）
（3）关联的列不要改名，要不update或delete操作找不到对应列，导致执行不成功

NEWDATA(sqlfunction=(select a.id as id,a.name as name,b.dept as dept from ds.t1 a, ds.t2 b where a.id=b.id and a.id = #id#))

五、加工通道参数一览表