目录
最近一段时间在做数据治理的数据中台项目，前两天项目上线。整个数据治理的流程是先从数据归集到数据清洗，再到数据转换，最后是数据质量。调度平台使用DS海豚，数仓没有选择Hadoop，而是选择了StarRocks。技术选型及整体架构，今天不讲，后面再专门写专栏。今天主要记录下数据归集时候遇到的问题。
项目中数据归集这个模块主要支持kafka/file/mysql/oracle/ocanBase等关系数据库的抽取，对于kafka和file类型任务交给DS，数据抽取的能力是直接使用StarRocks的sreamLoad组件来做的，我们做了加工，这里不展开讲。对于非kafka、file的数据库数据归集，主要使用了DataX。需要进行归集的库有两个，其中有一个库非常顺利完成了归集任务；另外一个库的三张表却全部失败。
通过给定的用户名、密码使用工具可以正常连接数据库，但是使用dataX连接失败，这个问题比较容易定位，通过查看源库的数据库版本，发现是mysql8.0的版本，而DataX3.0默认关于mysql的驱动版本是5.1.47，这个版本是不支持mysql8.0的。 通过修改mysql驱动即可解决问题。
这里要注意：
（1）不能直接替换原来的mysql-connector-java-j5.1.47.jar，直接替换是不生效的
（2）需要分别重新编译mysql的reader和writer，参考这里
（3）重新编译好的插件放到datax的plugin目录即可
完成上述问题之后，可以正常连接mysql了进行归集任务了
归集任务最后看是跑成功了，但是看细节日志发现，跑了18w的数据，有14w多都是失败，这实际上还是失败了。看了下错误日志，最多的日志就是报Too Many Versions，如下图所示：
首先，这个错误是在writer到StarRocks的时候，SR报的错误，关于这个错误，StarRocks的挂网是这样说的:
为什么会发生 "close index channel failed" 和 "too many tablet versions" 错误？应该如何处理？
上述报错是因为导入频率太快，数据没能及时合并 (Compaction) ，从而导致版本数超过支持的最大未合并版本数。默认支持的最大未合并版本数为 1000。可以通过如下方法解决上述报错：
增大单次导入的数据量，降低导入频率。
在 BE 的配置文件 be.conf 中修改以下配置，通过调整合并策略实现加快合并的目的：
cumulative_compaction_num_threads_per_disk = 4 base_compaction_num_threads_per_disk = 2 cumulative_compaction_check_interval_seconds = 2修改完成后，需要观察内存和 I/O，确保内存和 I/O 正常。
可以简单理解为，出现这种问题，是因为导入的频率太快了，可以通过修改导入的频次或者调整合并策略。但是Datax默认就是按照批次导入的，我们设置的是最大一次可以5w条数据，为什么还会出现这种问题？
仔细查看日志看到如下日志记录：
原来datax只要出现脏数据，插入失败的话，就会将这一批次的数据全部回滚，然后单条一条一条执行，这也是为什么只有18w的数据，却同步了将近2个小时，而且有14w多都是失败的原因了。既然是这样，那问题就还是得尝试去找脏数据，到底是哪些数据造成了失败。
PS：我们也尝试去调整SR的PE合并策略，但是效果不佳，而且这也不是解决问题的正确方法，毕竟数据在源库可以正常存储，就没有道理不能同步到数仓，即便真的是脏数据，也应该是在数仓的数据清洗流程对数据进行清洗。
通过继续查看日志，发现非常多如下错误：
这种错误看上去很诡异，通过对源表数据的分析，发现有一个字段的数据存在英文单引号，如下：
这个英文单引号是一个关键字符，datax在拼接insert into 语句的时候，字段类型如果是varchar或者char类型值是通过单引号包裹的，但是因为值中已经有一个单引号，这就造成识别错误，解析失败了。针对这个问题，我们尝试修改reader中的select语句，通过给字段添加replace来解决该问题，比如：
DataX的json文件配置根据数据源不同，配置不同，项目中是通过代码生成的，源表的colum是事先查询出来，在生成json文件的时候，循环生成querySql的，如下：
if("reader".equals(rwFlag) && ("varchar".equals(columm.getDataType()) ||
"char".equals(columm.getDataType()) ||
"string".equals(columm.getDataType()))){
sql.append("\"replace(`"+columm.getColumnName() + "`,\\\"'\\\",\\\"\\\")\"");
}else{
sql.append("\"`"+columm.getColumnName() + "`\"");
}通过这种方式，最后问题解决，并且也可以完成数据的初步清洗。同理，如果有其他特殊字符，也可以通过这种方式解决，如果有多个特殊字符，可以通过replace套replace解决。
关于类型转换的问题是做异构数据源之间数据同步必须要考虑的问题。 我们本次的源库是MYSQL，目标库是StarRocks。DBA同事梳理了字段类型的映射关系，如下所示：
| OB(mysql租户)/MySQL | StarRocks |
| bool | bool |
| boolean | boolean |
| tinyint | tinyint |
| smallint | smallint |
| mediumint | int |
| int | int |
| integer | int |
| bigint | bigint |
| decimal | DECIMAL |
| numeric | DECIMAL |
| float | float |
| double | double |
| bit | bit |
| datetime | datetime |
| timestamp | datetime |
| date | date |
| time | time |
| year | year |
| char | char |
| varchar | varchar |
| tinyblob | 无 |
| blob | 无 |
| mediumblob | 无 |
| longblob | 无 |
| tinytext | 无 |
| text | 无 |
| mediumtext | 无 |
| longtext | 无 |
（1）对于无的，可以统一使用string
（2）mysql中的varchar(255)是255个字符，可以存255个汉字或者英文，但是StarRocks中varchar(255)是255个字节，如果存中文是255/3个中文，所以在StarRocks中创建目标表的时候，要注意对应字段的长度为源库表中对应字段长度的3倍
（3）StraRocks中的string等价于VARCHAR(65535字节数)
（4）对于超过65535字节的字段长度可以进行截取、过滤，一般也不建议存长字段
按照上述方式处理之后，后面还是有一个问题，如下所示：
Error: Decimal '18392594342' is out of range. The type of 'ORDER_ACCOUNT' is DECIMAL64(10, 0)'. Row:
这个提示主要还是字段长度对不上，经过核查发现，主要问题在于取源表的数据结构时候，没有取到源表decimal类型的长度，SR中建表时候就使用了默认长度（10,0)，而恰巧源表中有更长的数据。最后，修改目标表该字段的长度即可，如下：
对应代码中，做了如下改造：
if (StringUtils.isNotBlank(colList.getColumnLength()) && !colList.getColumnType().equals("string")){
//字段类型+长度
sql1.append(" " + colList.getColumnType()+"("+colList.getColumnLength()+")" + " ");
} else if("decimal".equals(colList.getColumnType().toLowerCase())){
//如果是decimal类型，默认设置长度为(20,0)
sql1.append(" " + colList.getColumnType()+"(20,0)" + " ");
}
else {
//字段类型
sql1.append(" " + colList.getColumnType() + " ");
}完成上述调整后，问题解决。
这个问题比较隐蔽，在开发初期也没有留意，最近项目快要上线了，数据组的同事在做数据核查，发现同样查询条件的语句在业务源表上查询出来的数据条数和在数仓中归集过来的表查询条数不一样。
感觉百思不得其解，先尝试将数仓中同样查询条件的数据导出来再导入mysql，然后联合查询，看看到底差异数据是哪些，联合查询之后发现差异数据更多。 主要原因在于查询语句的条件是根据业务源表的一个create_time字段查询的。
最后直接取了某一条记录做对比，发现主键相同的两条记录，create_time不同。如下：
看到这里，猜测应该是时区的问题。 回过头去查看Datax的Json配置文件，发现reader和writer中的jdbcUrl地址都没有设置时区，加上时区，再次执行，问题解决。 参考这里。
系统已经上线有段时间了，今天早上监控平台突然连续报警，提示归集任务失败，看了下所有归集任务都失败了，仔细核查了发现是所有datax的任务都失败了，失败日志信息如下：
[INFO] 2022-12-02 00:00:02.414 +0800 [taskAppId=TASK-20221202-7705210273312_1-29152-87296] TaskLogLogger-class org.apache.dolphinscheduler.plugin.task.datax.DataxTask:[63] - ->
sudo: pam_open_session: System error
sudo: policy plugin failed session initialization出现这个问题的主要原因是因为启动datax的linux用户有密码有效期，刚好昨晚上密码到期了，然后再执行任务的时候，就报如上错误。 联系集成修改启动用户不适用sudo，避免类似问题发生，然后重新执行任务，任务执行成功。
在处理该问题的过程中，在修改了驱动后，还是报错时，当时怀疑是因为源表的字符集是utf8mb4，而StarRocks中没有utf8mb4的字符集，也不支持建表指定字符集，当时想，这样的话，就无解了。后来通过调研发现，并不是这样的问题，参考这里。