技术实践(1)-千万+数据量级别的数据清洗思路

技术实践(1)-千万+数据量级别的数据清洗思路

TianXinCoord

于 2020-08-10 14:18:22 发布

阅读量2.8k

收藏

4

点赞数

CC 4.0 BY-SA版权

分类专栏：

35_技术实践

文章标签：

大批量数据清洗

数据清洗

本文为博主原创文章，转载请注明出处！

本文链接：https://blog.csdn.net/sinat_34104446/article/details/107902694

35_技术实践

专栏收录该内容

5 篇文章

订阅专栏

本文探讨了大数据量级数据清洗的思路，包括不中断服务、中断服务及混合清洗方式，详细介绍了清洗步骤与多线程处理技巧，适用于千万级别业务数据。

摘要生成于

C知道

，由 DeepSeek-R1 满血版支持，

前往体验 >

一、文章说明

本篇博客主要是大数据量级别的数据清洗思路，因为使用的技术和数据库等存在差异，并没有提供具体的示例，但是各个场景思路基本一致完成清洗的方式很多，业务数据体量不同，本文可以作为其中千万级别业务数据清洗的一种思路，一般来说一个小时可以清洗几百万+的数据，具体速度数据库和配置的同步会有比较大的差异

二、文章正文

2.1 清洗背景

一些时候，因为业务上的需求变化或者某些不可变因素需要将数据库的数据进行处理，例如很常见的一个就是数据库敏感信息加密在项目初期，数据量不大的情况下可能我们存储的都是明文，但是当数据量到达百万和千万级别的时候，很多安全问题就开始暴露了，比如敏感信息需要加密等，或者就是数据采集系统的数据清洗也可能是依赖的业务系统需求有变化，要求对数据进行清洗，完成某些数据的转换，比如数据采集后对数据进行清洗整理

2.2 数据清洗方式

2.2.1 不中断服务进行清洗

不中断服务清洗指的是在清洗过程中不用停止服务完成清洗功能不中断服务进行清洗的优点就是：

a. 服务正常提供，不影响正常业务系统

b. 但是需要考虑数据清洗对数据库带来的压力，可能会导致正常业务系统变得缓慢

c. 不过这个都可以通过配置进行解决，具体参考后面内容

2.2.2 中断服务进行清洗

中断服务就是在月黑风高的夜晚(白天中断服务估计会被用户打死)悄悄把所有服务停掉，然后开启清洗服务，完成后神不知鬼不觉的启动服务中断服务清洗优点就是：

a. 可以把所有资源全部用于清洗上面，让清洗速度从单车到摩托，哈哈

b. 缺点就是需要确保在晚上的时候能够完成清洗，不要影响第二天的使用

2.2.3 不中断服务+中断服务阶段清洗

这种清洗方式是上面两种的综合体，具体操作步骤就是：

a. 指定一个时间为界限，然后之前的数据开始不中断服务进行清洗

b. 然后等到晚上再中断服务，清洗界限之后的数据好处就是可以把控整体的清洗进度

2.3 清洗步骤

2.3.1 前期准备

备份原始数据，这一步比较推荐做，不怕一万就怕万一将需要清洗的数据复制一份到中间表，到中间表的好处就是不影响正常业务表的使用，可以随时处理错误的数据，也可以不用中间表，不过这个就直接影响到原始业务数据，所以可以使用中建表，把清洗后的数据保存到此中间表中，数据检查没有问题之后再更新到原始业务表数据备份主要注意以下几方面

a. 备份数据采用的方式：如果数据不多，可以采用insert into xxx select xxx 的形式，直接查询出需要的数据保存到中间表中，如果数据量较大，也可采用分段导出再导入方式

b. 对于中间表，只保存需要的业务数据，不需要的字段不要放到里面注意如果使用中间表一定要给中间表创建索引， 有索引和没有索引速度会相差好几倍，所以一定要对更新中取数等条件添加必要所以

2.3.2 中期准备

这一步主要是完成代码的清洗，下面将以springboot项目为例清洗代码一定要使用多线程，例如1个线程1s可以清洗几百个，此时我可以开几百个线程同时进行清洗，但是一定要注意以下几个问题

a. 不要每个线程都去数据库取待清洗的数据，而是由一个主线程负责取数，然后分发到子线程进行处理，避免重复清洗数据

b. 例如现在由主线程取出1w待清洗数据，假如现在有100个子线程，那么就由主线程将10000/线程数个数据分给对应线程去清洗

c. 每个线程更新数据的条件一定不能重复，避免出现死锁，例如现在有联系人信息，我们可以通过数据行id或者身份证这种唯一标识的去更新数据，不要使用可能重复的数据去更新，这样每个线程更新都是行级锁，不会升级到表锁清洗中间表使用清洗状态字段，比如增加clean_status状态，每个值对应不同的清洗状态，这样方便重试代码或者手工处理更新一定是批量更新而不是单个更新，比如现在每个线程处理100个数据，每个更新完成后用list存储起来，等一百各清洗完成再做一次批量更新，减少数据库的交互次数，提高清洗速度对批量更新要做兼容处理，比如现在对100做批量更新可能因为一条数据问题导致100条更新失败，此时要对这个失败进行catch处理，然后再转为单行更新方式，最终只会失败1条，99条会成功，速度虽然慢，但是增加了健壮性多线程线程池的大小做成可配置类型，例如不同的时间段可以分配不同的线程数量和取数代销，下面是参考示例

a. 比如不中断服务就可以动态调节这里的线程大小和取数大小，调整到不会影响正常业务的合适大小，让他慢慢清洗就行，不过一定要注意分下来的一定是核心线程池大小，不是最大线程池大小，分了多少个线程，就需要多少个core线程同时运行！

clean:

# 单次取数大小

pageSize: 10000

# 队列大小

queueSize: 100

# 线程池的大小就等于：(pageSize / queueSize) + 1

2.3.3 后期准备

这一步主要就是检查了清洗数据没有问题之后，将中间表数据更新到原始表中，然后删除中间表根据业务不同也不需要用业务表，比如直接取出来清洗完以后可以直接更新原表，或者就是清洗之后直接就存储到另外的表中或者库中然后就是检查更新回去的值是否正确，都没有问题删除原始备份表