java之CSV大批量數(shù)據(jù)入庫的實現(xiàn)
更新時間：2023年02月05日 11:11:01 作者：邱可愛
本文主要介紹了java之CSV大批量數(shù)據(jù)入庫的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
需求
讀200+的CSV/EXCEL文件，按文件名稱存到不同數(shù)據(jù)庫
前期準備
環(huán)境
maven + jdk8 + mysql
代碼展示
pom文件
<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>
關鍵代碼及思路
多線程處理數(shù)據(jù)，否則8k萬數(shù)據(jù)太慢了
創(chuàng)建線程池，個數(shù)的大小一般取決于自己電腦配置，以及I/O還是CPU密集型。
- I/O密集型：CPU*2 + 1
- CPU密集型：CPU + 1
- Java查看CPU數(shù)目
public static final Integer N_Thread = Runtime.getRuntime().availableProcessors() * 2 + 1;
ExecutorService service = Executors.newFixedThreadPool(N_Thread ); List<CompletableFuture<Void>> futureList = new ArrayList<>();
數(shù)據(jù)進行分批處理，如果數(shù)據(jù)一次過大，可能導致sql拼接失敗或者程序連接超時的問題。
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);
// 加入線程集合，方便后續(xù)阻塞主線程，防止線程沒跑就提前結(jié)束
futureList.add(completableFuture);
basic++;
}while (basic * 6000 < total);
獲取文件數(shù)據(jù)
String url = dataConfig.getStation();
List<String> stationFiles = FileUtils.findFiles(url);
//遍歷文件
for (String fileName : stationFiles) {
try {
// 讀取文件數(shù)據(jù)
String path = url.concat(fileName);
CSVReader reader = new CSVReader(new InputStreamReader(new FileInputStream(path), StandardCharsets.UTF_8));
List<String[]> lines = reader.readAll();
根據(jù)文件名創(chuàng)建相應數(shù)據(jù)庫
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);
}
注意：如果在主線程完成之后，沒有對主線程進行阻塞，會導致線程池中的線程沒跑完就直接結(jié)束了，需要遍歷線程集合來阻塞主線程
for (CompletableFuture<Void> future : futureList) {
future.join();
}
關鍵的SQL語句
saveOrUpdateBatch
這里選用Mysql提供的ON DUPLICATE KEY UPDATE來實現(xiàn)更新或者插入，如果primary key 或者 unique key不存在就插入，否則就更新。注意：primary key 和 unique key都存在的時候可能會導致數(shù)據(jù)的更新的異常，這里建議選其中一個最為鍵，否則容易死鎖！見方案選型
<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+的數(shù)據(jù)；
在程序中查詢數(shù)據(jù)庫中現(xiàn)存的數(shù)據(jù)，然后對這些數(shù)據(jù)進行更新，其余的進行插入
此方案程序代碼看起來較繁瑣。
選用 ON DUPLICATE KEY UPDATE & 多線程來實現(xiàn)批量處理。 問題 在處理最后一些數(shù)據(jù)時，報異常：獲取不到數(shù)據(jù)庫連接，連接超時
解決方案
修改application.yml的sql配置
hikari:
connection-timeout: 600000 //時間設的長一些
當即存在primary key 以及 unique key時，出現(xiàn)了死鎖
如果一個表定義有多個唯一鍵（包括唯一索引、主鍵）時，是不安全的。
當mysql執(zhí)行INSERT ON DUPLICATE KEY的INSERT時，存儲引擎會檢查插入的行為是否產(chǎn)生重復錯誤。
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重新建表。
批量數(shù)據(jù)入庫，當SQL語句拼接過長，超過了設置的最大的限制。
解決方案
批量數(shù)據(jù)入庫時，稍微減少數(shù)據(jù)量再進行插入，如6000條數(shù)據(jù)減為4000數(shù)據(jù)之后再批量入庫。
運行結(jié)果
相比之前一天200W+數(shù)據(jù)有了質(zhì)的提升，半小時完成了所有數(shù)據(jù)的預處理以及入庫。
到此這篇關于java之CSV大批量數(shù)據(jù)入庫的實現(xiàn)的文章就介紹到這了,更多相關java CSV大批量入庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
相關文章
這篇文章主要為大家詳細介紹了JavaScript實現(xiàn)鼠標移動粒子跟隨效果，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-08-08
JSON(JavaScript?Object?Notation)是一種輕量級的數(shù)據(jù)交換格式，它算是JavaScript語言的一部分，與XML一樣都可以用于數(shù)據(jù)的存儲和傳輸，本文講給大家介紹如何進行JSON解析,需要的朋友可以參考下2023-12-12
這篇文章主要介紹了淺談Redis持久化的幾種方式,前面說到了Redis持久化的 實現(xiàn)方式主要分為了：快照持久化(RDB)、寫日志持久化(AOF)2023-08-08
,其中快照持久化方式也就是RDB ,需要的朋友可以參考下
java使用Hex編碼解碼實現(xiàn)Aes加密解密功能示例
這篇文章主要介紹了java使用Hex編碼解碼實現(xiàn)Aes加密解密功能,結(jié)合完整實例形式分析了Aes加密解密功能的定義與使用方法,需要的朋友可以參考下2017-01-01
Spring中事務管理的四種方法（銀行轉(zhuǎn)賬為例）
這篇文章主要給大家介紹了關于Spring中事務管理的四種方法，文中是以銀行轉(zhuǎn)賬為例，通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-05-05
這篇文章主要為大家詳細介紹了java實現(xiàn)簡單超市管理系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-01-01
Spring Boot 參數(shù)校驗的具體實現(xiàn)方式
這篇文章主要介紹了Spring Boot 參數(shù)校驗的具體實現(xiàn)方式，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2019-06-06
本文將詳細介紹redisson實現(xiàn)分布式鎖原理。具有很好的參考價值，下面跟著小編一起來看下吧2017-02-02
利用Intellij Idea連接遠程服務器實現(xiàn)遠程上傳部署功能
大家在使用Intellij Idea開發(fā)程序的時候，是不是需要部署到遠程SSH服務器運行呢，當然也可以直接在idea軟件內(nèi)容實現(xiàn)配置部署操作，接下來通過本文給大家分享利用Intellij Idea連接遠程服務器實現(xiàn)遠程上傳部署功能，感興趣的朋友跟隨小編一起看看吧2021-05-05
今天小編就為大家分享一篇關于Java面向?qū)ο筮x擇題總結(jié)歸納，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-01-01