30G 上亿数据的超大文件,如何快速导入生产环境?
Hello,大家好,我是楼下小黑哥~
如果给你一个包含一亿行数据的超大文件,让你在一周之内将数据转化导入生产数据库,你会如何操作?
上面的问题其实是小黑哥前段时间接到一个真实的业务需求,将一个老系统历史数据通过线下文件的方式迁移到新的生产系统。
由于老板们已经敲定了新系统上线时间,所以只留给小黑哥一周的时间将历史数据导入生产系统。
由于时间紧,而数据量又超大,所以小黑哥设计的过程想到一下解决办法:
- 拆分文件
- 多线程导入
欢迎关注我的公众号:小黑十一点半,获得日常干货推送。如果您对我的专题内容感兴趣,也可以关注我的博客:studyidea.cn
拆分文件
首先我们可以写个小程序,或者使用拆分命令 split
将这个超大文件拆分一个个小文件。
-- 将一个大文件拆分成若干个小文件,每个文件 100000 行
split -l 100000 largeFile.txt -d -a 4 smallFile_
这里之所以选择先将大文件拆分,主要考虑到两个原因:
第一如果程序直接读取这个大文件,假设读取一半的时候,程序突然宕机,这样就会直接丢失文件读取的进度,又需要重新开头读取。
而文件拆分之后,一旦小文件读取结束,我们可以将小文件移动一个指定文件夹。
这样即使应用程序宕机重启,我们重新读取时,只需要读取剩余的文件。
第二,一个文件,只能被一个应用程序读取,这样就限制了导入的速度。
而文件拆分之后,我们可以采用多节点部署的方式,水平扩展。每个节点读取一部分文件,这样就可以成倍的加快导入速度。
多线程导入
当我们拆分完文件,接着我们就需要读取文件内容,进行导入。
之前拆分的时候,设置每个小文件包含 10w 行的数据。由于担心一下子将 10w 数据读取应用中,导致堆内存占用过高,引起频繁的 Full GC,所以下面采用流式读取的方式,一行一行的读取数据。
当然了,如果拆分之后文件很小,或者说应用的堆内存设置很大,我们可以直接将文件加载到应用内存中处理。这样相对来说简单一点。
逐行读取的代码如下:
File file = ...
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line=iterator.nextLine();convertToDB(line);}}
上面代码使用 commons-io
中的 LineIterator
类,这个类底层使用了 BufferedReader
读取文件内容。它将其封装成迭代器模式,这样我们可以很方便的迭代读取。
如果当前使用 JDK1.8 ,那么上述操作更加简单,我们可以直接使用 JDK 原生的类 Files
将文件转成 Stream
方式读取,代码如下:
Files.lines(Paths.get("文件路径"), Charset.defaultCharset()).forEach(line -> {convertToDB(line);
});
其实仔细看下 Files#lines
底层源码,其实原理跟上面的 LineIterator
类似,同样也是封装成迭代器模式。
多线程的引入存在的问题
上述读取的代码写起来不难,但是存在效率问题,主要是因为只有单线程在导入,上一行数据导入完成之后,才能继续操作下一行。
为了加快导入速度,那我们就多来几个线程,并发导入。
多线程我们自然将会使用线程池的方式,相关代码改造如下:
File file = ...;
ExecutorService executorService = new ThreadPoolExecutor(5,10,60,TimeUnit.MINUTES,// 文件数量,假设文件包含 10W 行new ArrayBlockingQueue<>(10*10000),// guava 提供new ThreadFactoryBuilder().setNameFormat("test-%d").build());
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line = iterator.nextLine();executorService.submit(() -> {convertToDB(line);});}}
上述代码中,每读取到一行内容,就会直接交给线程池来执行。
我们知道线程池原理如下:
- 如果核心线程数未满,将会直接创建线程执行任务。
- 如果核心线程数已满,将会把任务放入到队列中。
- 如果队列已满,将会再创建线程执行任务。
- 如果最大线程数已满,队列也已满,那么将会执行拒绝策略。
由于我们上述线程池设置的核心线程数为 5,很快就到达了最大核心线程数,后续任务只能被加入队列。
为了后续任务不被线程池拒绝,我们可以采用如下方案:
- 将队列容量设置成很大,包含整个文件所有行数
- 将最大线程数设置成很大,数量大于件所有行数
以上两种方案都存在同样的问题,第一种是相当于将文件所有内容加载到内存,将会占用过多内存。
而第二种创建过多的线程,同样也会占用过多内存。
一旦内存占用过多,GC 无法清理,就可能会引起频繁的 Full GC,甚至导致 OOM,导致程序导入速度过慢。
解决这个问题,我们可以如下两种解决方案:
CountDownLatch
批量执行- 扩展线程池
CountDownLatch
批量执行
JDK 提供的 CountDownLatch
,可以让主线程等待子线程都执行完成之后,再继续往下执行。
利用这个特性,我们可以改造多线程导入的代码,主体逻辑如下:
try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {// 存储每个任务执行的行数List<String> lines = Lists.newArrayList();// 存储异步任务List<ConvertTask> tasks = Lists.newArrayList();while (iterator.hasNext()) {String line = iterator.nextLine();lines.add(line);// 设置每个线程执行的行数if (lines.size() == 1000) {// 新建异步任务,注意这里需要创建一个 Listtasks.add(new ConvertTask(Lists.newArrayList(lines)));lines.clear();}if (tasks.size() == 10) {asyncBatchExecuteTask(tasks);}}// 文件读取结束,但是可能还存在未被内容tasks.add(new ConvertTask(Lists.newArrayList(lines)));// 最后再执行一次asyncBatchExecuteTask(tasks);
}
这段代码中,每个异步任务将会导入 1000 行数据,等积累了 10 个异步任务,然后将会调用 asyncBatchExecuteTask
使用线程池异步执行。
/*** 批量执行任务** @param tasks*/
private static void asyncBatchExecuteTask(List<ConvertTask> tasks) throws InterruptedException {CountDownLatch countDownLatch = new CountDownLatch(tasks.size());for (ConvertTask task : tasks) {task.setCountDownLatch(countDownLatch);executorService.submit(task);}// 主线程等待异步线程 countDownLatch 执行结束countDownLatch.await();// 清空,重新添加任务tasks.clear();
}
asyncBatchExecuteTask
方法内将会创建 CountDownLatch
,然后主线程内调用 await
方法等待所有异步线程执行结束。
ConvertTask
异步任务逻辑如下:
/*** 异步任务* 等数据导入完成之后,一定要调用 countDownLatch.countDown()* 不然,这个主线程将会被阻塞,*/
private static class ConvertTask implements Runnable {private CountDownLatch countDownLatch;private List<String> lines;public ConvertTask(List<String> lines) {this.lines = lines;}public void setCountDownLatch(CountDownLatch countDownLatch) {this.countDownLatch = countDownLatch;}@Overridepublic void run() {try {for (String line : lines) {convertToDB(line);}} finally {countDownLatch.countDown();}}
}
ConvertTask
任务类逻辑就非常简单,遍历所有行,将其导入到数据库中。所有数据导入结束,调用 countDownLatch#countDown
。
一旦所有异步线程执行结束,调用 countDownLatch#countDown
,主线程将会被唤醒,继续执行文件读取。
虽然这种方式解决上述问题,但是这种方式,每次都需要积累一定任务数才能开始异步执行所有任务。
另外每次都需要等待所有任务执行结束之后,才能开始下一批任务,批量执行消耗的时间等于最慢的异步任务消耗的时间。
这种方式线程池中线程存在一定的闲置时间,那有没有办法一直压榨线程池,让它一直在干活呢?
扩展线程池
回到最开始的问题,文件读取导入,其实就是一个生产者-消费者消费模型。
主线程作为生产者不断读取文件,然后将其放置到队列中。
异步线程作为消费者不断从队列中读取内容,导入到数据库中。
一旦队列满载,生产者应该阻塞,直到消费者消费任务。
其实我们使用线程池的也是一个生产者-消费者消费模型,其也使用阻塞队列。
那为什么线程池在队列满载的时候,不发生阻塞?
这是因为线程池内部使用 offer
方法,这个方法在队列满载的时候不会发生阻塞,而是直接返回 。
那我们有没有办法在线程池队列满载的时候,阻塞主线程添加任务?
其实是可以的,我们自定义线程池拒绝策略,当队列满时改为调用 BlockingQueue.put
来实现生产者的阻塞。
RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {if (!executor.isShutdown()) {try {executor.getQueue().put(r);} catch (InterruptedException e) {// should not be interrupted}}}
};
这样一旦线程池满载,主线程将会被阻塞。
使用这种方式之后,我们可以直接使用上面提到的多线程导入的代码。
ExecutorService executorService = new ThreadPoolExecutor(5,10,60,TimeUnit.MINUTES,new ArrayBlockingQueue<>(100),new ThreadFactoryBuilder().setNameFormat("test-%d").build(),(r, executor) -> {if (!executor.isShutdown()) {try {// 主线程将会被阻塞executor.getQueue().put(r);} catch (InterruptedException e) {// should not be interrupted}}});
File file = new File("文件路径");try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream(file), "UTF-8")) {while (iterator.hasNext()) {String line = iterator.nextLine();executorService.submit(() -> convertToDB(line));}
}
小结
一个超大的文件,我们可以采用拆分文件的方式,将其拆分成多份文件,然后部署多个应用程序提高读取速度。
另外读取过程我们还可以使用多线程的方式并发导入,不过我们需要注意线程池满载之后,将会拒绝后续任务。
我们可以通过扩展线程池,自定义拒绝策略,使读取主线程阻塞。
好了,今天文章内容就到这里,不知道各位有没有其他更好的解决办法,欢迎留言讨论。
欢迎关注我的公众号:小黑十一点半,获得日常干货推送。如果您对我的专题内容感兴趣,也可以关注我的博客:studyidea.cn
30G 上亿数据的超大文件,如何快速导入生产环境?相关推荐
- php 百度网盘上传文件大小限制吗,百度云存储,curl_百度云存储如何上传数G超大文件(视频类)和防盗链等一系列问题,百度云存储,curl - phpStudy...
百度云存储如何上传数G超大文件(视频类)和防盗链等一系列问题 最近在开发一个视频网站,打算用百度开放云平台上的云存储,但视频较大,一般有几个G(高清电影).1G一下的视频管理控制台里能用页面上传,但1 ...
- 前端如何展示一个有上亿数据的树结构
需求描述 数据上亿,有数千个从属于根节点的一级子节点: 每个一级子节点有数个到数万个二级子节点: 二级子节点可能有三级子节点: 子节点的最多层深不定(即,也许可能有 10 层): 需求分析 当面临一个 ...
- 耗时3天,上亿数据如何做到秒级查询?
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试文章 来源:sohu.gg/jIp59N 最近在忙着优化集团公司的一个报 ...
- 查询a表有但是b表没有的数据_牛逼!一个上亿数据的报表竟然能做到秒查~
数据背景 首先项目是西门子中国在我司实施部署的MES项目,由于项目是在产线上运作(3 years+),数据累积很大.在项目的数据库中,大概上亿条数据的表有5个以上,千万级数据的表10个以上,百万级数据 ...
- 耗时 3 天,上亿数据如何做到秒级查询?
最近在忙着优化集团公司的一个报表.优化完成后,报表查询速度由从半小时以上(甚至查不出)到秒查的质变.从修改 SQL 查询语句逻辑到决定创建存储过程实现,花了我 3 天多的时间,在此总结一下,希望对朋友 ...
- 实战上亿数据,如何实现秒查!
点击上方 好好学java ,选择 星标 公众号 重磅资讯.干货,第一时间送达 今日推荐:硬刚一周,3W字总结,一年的经验告诉你如何准备校招! 个人原创100W+访问量博客:点击前往,查看更多 作者:J ...
- 数据泵impdp上亿数据量的生产环境导入参考(完全本人总结,与公司机密无关)
总述: 目前动不动几十年的数据量迁移,10年前的exp/imp已经成为历史,对于1G以内的expdp就不用看此文了,对于很多大行业名声数据库相关的历史数据迁移,往往都是100G以上的dmp,而且做过i ...
- 上亿数据怎么玩深度分页?兼容MySQL + ES + MongoDB
面试题 & 真实经历 面试题:在数据量很大的情况下,怎么实现深度分页? 大家在面试时,或者准备面试中可能会遇到上述的问题,大多的回答基本上是分库分表建索引,这是一种很标准的正确回答,但现实总是 ...
- 实战:上亿数据如何秒查
作者:blog.csdn.net/chenleixing/article/details/44994571 最近在忙着优化集团公司的一个报表.优化完成后,报表查询速度有从半小时以上(甚至查不出)到秒查 ...
- 高效率批量插入上亿数据
转至 : https://blog.csdn.net/xiyang_1990/article/details/78771962 这里提供一种方法,使用 APPEND 提示,使得十分钟内插入上亿数据成为 ...
最新文章
- 不用任何第三方,写一个RTMP直播推流器
- U-Boot的LDS文件分析
- GMIS 2017大会戴文渊演讲:构建AI商业大脑
- hash地址_到底什么是Hash?
- 解决 Windows Update 时提示当前无法检查更新,因为未运行服务
- 我从草原来:自由摄影人李伟 (内蒙古电视台“蔚蓝的故乡”20110407)
- React Native之(var和let区别 )(简单解构)(map对象遍历)(可变顺序参数和不可以变顺序参数函数)
- 二维数组常用的赋值方式
- 复信号两种不同加噪方式的比较
- SQL 语句 - Select(1): 指定表
- 【英语学习】【WOTD】gullible 释义/词源/示例
- zookeeper版本更新_很遗憾,没有一篇文章能讲清楚ZooKeeper
- 如何选择适合你的兴趣爱好(四十六),豫剧
- 计算机与调制解调器之间接口为,单片机串行接口和调制解调器如何实现远程通信...
- 对JavaScript中 this 关键词的理解
- U盘不能复制超过4G的文件解决方法
- 信息系统项目管理师(第1章)
- java espresso_java – 带有片段的Android Espresso功能测试
- 云享·人物丨造梦、探梦、筑梦,三位开发者在华为云上的寻梦之旅
- 5个增加设计趣味性的方法
热门文章
- 二维树状数组 探索进行中
- OpenJudge计算概论-字符串最大跨距
- Iframe和母版页(.net)
- 查找整数的位置c语言,1-1 编程总结:查找整数
- 预处理_关于食材预处理
- 拓端tecdat|R语言法国足球联赛球员多重对应分析(MCA)
- Java EE实战教程 servlet (一)
- vs code搭建vue环境
- Ubuntu[16.04/18.04/20.04] arm 下修改本地源 sources.list 为国内镜像
- 服务器真在运行中,win7电脑服务器正在运行中的解决教程