本站首页 收藏本站
广告位联系

APP正在开发中...
返回顶部
您的位置： 主页 > 软件编程 > java >

收藏
分享到

java之CSV大批量数据入库的实现方法

java 来源：未知 作者：佚名 发布时间：2023-11-08 23:18:04 人浏览
摘要

需求 读200+的CSV/EXCEL文件，按文件名称存到不同数据库 前期准备 环境 maven + jdk8 + mysql 代码展示 pom文件 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

需求

读200+的CSV/EXCEL文件，按文件名称存到不同数据库

前期准备

环境

maven + jdk8 + mysql

代码展示

pom文件

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

<dependencies>

    <!--https://mvnrepository.com/artifact/com.opencsv/opencsv -->

    <dependency>            //读取csv文件

        <groupId>com.opencsv</groupId>

        <artifactId>opencsv</artifactId>

        <version>${opencsv.version}</version>

    </dependency>

 

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-web</artifactId>

    </dependency>

    <!--mybatis-plus 持久层-->

    <dependency>

        <groupId>com.baomidou</groupId>

        <artifactId>mybatis-plus-boot-starter</artifactId>

        <version>${mybatis-plus.version}</version>

    </dependency>

 

    <!-- velocity 模板引擎, Mybatis Plus 代码生成器需要 -->

    <dependency>

        <groupId>org.apache.velocity</groupId>

        <artifactId>velocity-engine-core</artifactId>

        <version>${velocity.version}</version>

    </dependency>

    <dependency>

        <groupId>mysql</groupId>

        <artifactId>mysql-connector-java</artifactId>

        <scope>runtime</scope>

    </dependency>

    <dependency>

        <groupId>org.projectlombok</groupId>

        <artifactId>lombok</artifactId>

        <optional>true</optional>

    </dependency>

    <dependency>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-test</artifactId>

        <scope>test</scope>

    </dependency>

</dependencies>

关键代码及思路

多线程处理数据，否则8k万数据太慢了

创建线程池，个数的大小一般取决于自己电脑配置，以及I/O还是CPU密集型。

  • I/O密集型：CPU*2 + 1
  • CPU密集型：CPU + 1
  • Java查看CPU数目

1

public static final Integer N_Thread = Runtime.getRuntime().availableProcessors() * 2 + 1;

1

2

ExecutorService service = Executors.newFixedThreadPool(N_Thread );

List<CompletableFuture<Void>> futureList = new ArrayList<>();

数据进行分批处理，如果数据一次过大，可能导致sql拼接失败或者程序连接超时的问题。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

List<AirStationHourEntity> records = new ArrayList<>(table.values());

int basic = 0, total = records.size();

 

do{

    int finalBasic = basic;

    String finalTableName = tableName;

    CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {

        List<AirStationHourEntity> subRecords = records.subList(finalBasic * 6000, Math.min((1 + finalBasic) * 6000, total));

        airStationHourMapper.saveOrUpdateBatch(finalTableName, subRecords);

    }, service);

    // 加入线程集合，方便后续阻塞主线程，防止线程没跑就提前结束

    futureList.add(completableFuture);

    basic++;

}while (basic * 6000 < total);

获取文件数据

1

2

3

4

5

6

7

8

9

String url = dataConfig.getStation();

List<String> stationFiles = FileUtils.findFiles(url);

//遍历文件

for (String fileName : stationFiles) {

    try {

        // 读取文件数据

        String path = url.concat(fileName);

        CSVReader reader = new CSVReader(new InputStreamReader(new FileInputStream(path), StandardCharsets.UTF_8));

        List<String[]> lines = reader.readAll();

根据文件名创建相应数据库

1

2

3

4

5

tableName = fileName.substring(fileName.lastIndexOf("_") + 1, fileName.indexOf(".") - 2);

               tableName = TABLE_PREFIX.STATION_PREFIX.value.concat(tableName);

               if (airStationHourMapper.findTableByName(tableName) == 0) {

                   airStationHourMapper.createNewTable(tableName);

               }

注意：如果在主线程完成之后，没有对主线程进行阻塞，会导致线程池中的线程没跑完就直接结束了，需要遍历线程集合来阻塞主线程

1

2

3

for (CompletableFuture<Void> future : futureList) {

    future.join();

}

关键的SQL语句

saveOrUpdateBatch

这里选用Mysql提供的ON DUPLICATE KEY UPDATE来实现更新或者插入，如果primary key 或者 unique key不存在就插入，否则就更新。注意：primary key 和 unique key都存在的时候可能会导致数据的更新的异常，这里建议选其中一个最为键，否则容易死锁！见方案选型

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

<update id="saveOrUpdateBatch">

       insert into ${tableName}

       (

       <include refid="Base_Column_List"/>

       )

       values

       <foreach collection="records" item="record" separator=",">

           (

           #{record.dataId,jdbcType=VARCHAR},

           #{record.stationId,jdbcType=VARCHAR},

           #{record.dataTime,jdbcType=TIMESTAMP},

           #{record.aqi,jdbcType=INTEGER},

           #{record.priPol,jdbcType=VARCHAR},

           #{record.co,jdbcType=DOUBLE},

           #{record.co24,jdbcType=DOUBLE},

           #{record.no2,jdbcType=INTEGER},

           #{record.no224,jdbcType=INTEGER},

           #{record.so2,jdbcType=INTEGER},

           #{record.so224,jdbcType=INTEGER},

           #{record.o3,jdbcType=INTEGER},

           #{record.o324,jdbcType=INTEGER},

           #{record.o38,jdbcType=INTEGER},

           #{record.o3824,jdbcType=INTEGER},

           #{record.pm10,jdbcType=INTEGER},

           #{record.pm1024,jdbcType=INTEGER},

           #{record.pm25,jdbcType=INTEGER},

           #{record.pm2524,jdbcType=INTEGER}

           )

       </foreach>

       ON DUPLICATE KEY UPDATE

       AQI = VALUES(AQI),pri_pol = VALUES(pri_pol),

       CO = VALUES(CO),CO_24 = VALUES(CO_24),

       NO2 = VALUES(NO2),NO2_24 = VALUES(NO2_24),

       SO2 = VALUES(SO2),SO2_24 = VALUES(SO2_24),

       O3 = VALUES(O3),O3_8 = VALUES(O3_8),O3_8_24 = VALUES(O3_8_24),

       PM10 = VALUES(PM10),PM10_24 = VALUES(PM10_24),

       PM2_5 = VALUES(PM2_5),PM2_5_24 =VALUES(PM2_5_24);

   </update>

方案选型

此方案时间过久，10多个小时大概能完成100w+的数据；

在程序中查询数据库中现存的数据，然后对这些数据进行更新，其余的进行插入

此方案程序代码看起来较繁琐。

选用 ON DUPLICATE KEY UPDATE & 多线程来实现批量处理。 问题 在处理最后一些数据时，报异常：获取不到数据库连接，连接超时

解决方案

修改application.yml的sql配置

1

2

hikari:

  connection-timeout: 600000  //时间设的长一些

当即存在primary key 以及 unique key时，出现了死锁

如果一个表定义有多个唯一键（包括唯一索引、主键）时，是不安全的。
当mysql执行INSERT ON DUPLICATE KEY的INSERT时，存储引擎会检查插入的行为是否产生重复错误。

1

2

3

4

5

6

7

org.springframework.dao.DeadlockLoserDataAccessException:

### Error updating database.  Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction

### The error may exist in class path resource [mapper/PersonGroupRefMapper.xml]

### The error may involve com.order.addOrder-Inline

### The error occurred while setting parameters

### SQL: insert into t_***(XX,XX,XX,XX,XX,XX) values         (?, ?, ?, ?, ?, ?)         ON DUPLICATE KEY UPDATE XX= VALUES(XX), XX= VALUES(XX), XX= VALUES(XX)

### Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction Deadlock found when trying to get lock; try restarting transaction; nested exception is com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transactio

解决方案

见上面关键SQL，删除了primary key 留下了unique key重新建表。

见上面关键SQL，删除了primary key 留下了unique key重新建表。

批量数据入库，当SQL语句拼接过长，超过了设置的最大的限制。

解决方案

批量数据入库时，稍微减少数据量再进行插入，如6000条数据减为4000数据之后再批量入库。

运行结果

相比之前一天200W+数据有了质的提升，半小时完成了所有数据的预处理以及入库。
版权声明 : 本文内容来源于互联网或用户自行发布贡献，该文观点仅代表原作者本人。本站仅提供信息存储空间服务和不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权， 违法违规的内容， 请发送邮件至2530232025#qq.cn(#换@)举报，一经查实，本站将立刻删除。

您可能感兴趣的文章 :

原文链接 :
相关文章

  • Java访问者模式实现优雅的对象结构处理

    Java访问者模式实现优雅的对象结构处理
    Java访问者模式(Visitor Pattern)是一种行为型设计模式,它允许将算法与其所操作的对象分离.该模式定义了一个访问者对象,它可以访问不同类型

  • java之CSV大批量数据入库的实现方法

    java之CSV大批量数据入库的实现方法
    需求 读200+的CSV/EXCEL文件，按文件名称存到不同数据库 前期准备 环境 maven + jdk8 + mysql 代码展示 pom文件 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

  • SpringBoot起步依赖和自动配置介绍

    SpringBoot起步依赖和自动配置介绍
    1、起步依赖 概念 起步依赖本质上是一个Maven项目对象模型（Project Object Model，POM），定义了对其他库的传递依赖，这些东西加在一起支持某

  • intellij idea自动生成类注释和方法注释配置方法

    intellij idea自动生成类注释和方法注释配置方法
    intellij idea自动生成类注释和方法注释 1.新建类，自动生成类注释的模板配置 File-Settings-Editor-File and Code Templates-Class 1 2 3 4 5 /** @Description: T
  • Java线程的停止实现原理介绍
    线程停止的原理 使用interrupt来通知，而不是强制 java提供了interrrupt让一个线程来通知另一个线程停止 如果想中断一个线程，但是那个线程不

  • springboot按月分表的实现方式介绍

    springboot按月分表的实现方式介绍
    在实际工作中，会遇到业务比较集中的情况，随着时间推延，这部分业务关联的mysql表就会越来越大，十分臃肿。尽管在项目架构上做了读写

  • jmap执行失败如何获取heapdump

    jmap执行失败如何获取heapdump
    在之前的OOM问题复盘中，我们添加了jmap脚本来自动dump内存现场，方便排查OOM问题。 但当我反复模拟OOM场景测试时，发现jmap有时可以dump成功

  • Spring refresh()源码解析介绍

    Spring refresh()源码解析介绍
    正文 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 public void refresh() throws BeansException, IllegalStateExcept
  • 线程阻塞唤醒工具LockSupport使用介绍
    LockSupport 简介 LockSupport是 Java 并发编程中一个非常重要的组件，我们熟知的并发组件Lock、线程池、CountDownLatch等都是基于AQS实现的，而AQS内
  • RocketMQ的存储模型全介绍
    RocketMQ有Producer、Consumer、NameSrv、Broker四个部分。其中Broker用于存储消息，维护消息队列和订阅关系，是RocketMQ四个部分中最重要的一个部分
推荐阅读
最新更新
友情链接
  • 本站所有内容来源于互联网或用户自行发布，本站仅提供信息存储空间服务，不拥有版权，不承担法律责任。如有侵犯您的权益，请您联系站长处理！
  • Copyright © 2017-2022 F11.CN All Rights Reserved. F11站长开发者网 版权所有 | 苏ICP备2022031554号-1 | 51LA统计