快速导入上亿行数据文件到数据库表(使用 JDBC 的 executeBatch)

最近在 cnblogs 网站上，看其他人博客，谈及一个包含很多行(一亿)的大文件，一周之内，将其数据导入到数据库表。我谈到可以使用"使用数据库事务，分批 commit 到数据库，每批次有 5000行"的方法，提高数据导入速度，两天应该就可以了。好像博主及下方评论者，不太理解，这个"分批 commit "。特写此博客，介绍一下使用 JDBC 的 executeBatch 做分批 commit，以提高大批量数据的导入速度。

最近在 cnblogs 网站上，看其他人博客，谈及一个包含很多行(一亿)的大文件，一周之内，将其数据导入到数据库表。

我谈到可以使用"使用数据库事务，分批 commit 到数据库，每批次有 5000行"的方法，提高数据导入速度，两天应该就可以了。

好像博主及下方评论者，不太理解，这个"分批 commit "。

特写此博客，介绍一下使用 JDBC 的 executeBatch 做分批 commit，以提高大批量数据的导入速度。

 

JDBC 有个 PreparedStatement 类，包含 addBatch, executeBatch 等函数(或称之为方法，我不区分这两个概念)。配合 Connection 的setAutoCommit(false), commit()，即可实现"分批 commit "。

当然，首先要逐行读数据文件。这里的数据文件，一般是 .txt 或 .csv 之类的纯文本文件，以逗号作为列分割，有的以 tab 做分割字符，也有的使用固定列宽(比如1-4字符为第一列，5-12为第二列...)。

我们使用 BufferedReader 来实现逐行读取。这是一个常用的 Java 类，可很好地用于此处文件读取。

为了方便起见，软件将从 Java 命令行读取 JVM 参数，举例如下:

-Ddata_file=C:\svn_projects\sgm_small_projects\batch_data_import\data\sample_data_1w.csv -Dfrom_line=1 -Dto_line= -Dbatch_commit_size=5000 -Duse_multi_thread=false

其中，

参数 data_file 为数据文件；

from_line 用于指定数据文件中的起始行号，最小值为1，一方面可用于跳过标题行，另一方面，可用于长时间运行过程中，如有中断，可重新从某行开始;

to_line 用于指定数据文件中的结束行号，可空;

batch_commit_size 用于指定每批次的数据行数，可调整，以便测试哪种参数，导入数据最快，此处配置为5000;

use_multi_thread 用于指定程序是否使用多线程，此参数暂无用处。

 

大批量数据文件导入，一般的策略为：

a. 正确的数据，尽量全部导入；

b. 错误的数据，跳过、记录报错数据行信息，继续运行；

c. 全部导入完成后，分析错误的数据，特殊处理。

 

以下介绍的代码，可以很好地实现这几个策略(报错到批次、行号范围)。运行时有类似如下的日志信息:

15:49:49.201 [main] INFO c.z.b.srv.DatabaseSrv - saveBatchDataInTrasaction begin,iBatchNum:1
15:49:51.416 [main] INFO c.z.b.srv.DatabaseSrv - 批量 commit,批次号:1, 原数据文件行[1-5000], 提交成功.
15:49:51.422 [main] INFO c.z.b.srv.DatabaseSrv - saveBatchDataInTrasaction begin,iBatchNum:2
15:49:52.306 [main] INFO c.z.b.srv.DatabaseSrv - 批量 commit,批次号:2, 原数据文件行[5001-10000], 提交成功.
15:49:52.329 [main] INFO c.z.b.srv.DatabaseSrv - saveBatchDataInTrasaction begin,iBatchNum:3
15:49:53.253 [main] INFO c.z.b.srv.DatabaseSrv - 批量 commit,批次号:3, 原数据文件行[10001-15000], 提交成功.
15:49:53.277 [main] INFO c.z.b.srv.DatabaseSrv - saveBatchDataInTrasaction begin,iBatchNum:4
15:49:54.188 [main] INFO c.z.b.srv.DatabaseSrv - 批量 commit,批次号:4, 原数据文件行[15001-20000], 提交成功.

 

主控程序 BatchDataImportMain ，功能为读取以上参数，然后使用 reader 读数据文件，最后调用 dataSrv.saveData ，代码如下:

public class BatchDataImportMain { public static void main(String[] args) {  Logger log = LoggerFactory.getLogger(BatchDataImportMain.class);  try {   log.info("从命令行参数中获取数据...");   String strDataFile = System.getProperty("data_file");   // 数据文件的第一行为1，不是0，方便用户理解   String strFromLine = System.getProperty("from_line", "1");   String strToLine = System.getProperty("to_line");   MutableObject<Long> iFromLine = null;   if (StringUtils.isNotEmpty(strFromLine)) {    iFromLine = new MutableObject<Long>();    iFromLine.setValue(Long.parseLong(strFromLine));   }   MutableObject<Long> iToLine = null;   if (StringUtils.isNotEmpty(strToLine)) {    iToLine = new MutableObject<Long>();    iToLine.setValue(Long.parseLong(strToLine));   }   String strBatchCommitSize = System.getProperty("batch_commit_size");   int iBatchCommitSize = 5000;   if (StringUtils.isNotEmpty(strBatchCommitSize)) {    iBatchCommitSize = Integer.parseInt(strBatchCommitSize);   }   String strUseMultiThread = System.getProperty("use_multi_thread");   boolean bUseMultiThread = false;   if (StringUtils.equalsIgnoreCase(strUseMultiThread, "true")) {    bUseMultiThread = true;   }   File fDataFile = new File(strDataFile);   log.info("begin save data from file:" + fDataFile.getAbsolutePath());   DataImportSrvBase dataSrv = null;   if (bUseMultiThread) {    dataSrv = new DataImportSrvUseThread();   } else {    dataSrv = new DataImportSrvNotUseThread();   }   try (FileInputStream fis = new FileInputStream(fDataFile)) {    String charsetName = "gbk";    try (InputStreamReader isr = new InputStreamReader(fis, charsetName)) {     try (BufferedReader br = new BufferedReader(isr)) {      dataSrv.saveData(br, iFromLine, iToLine, iBatchCommitSize, fDataFile.getName());     }    }   }   log.info("ends save data from file:" + fDataFile.getAbsolutePath());  } catch (Exception err) {   log.error(err.getMessage(), err);  } }}

 

以上 dataSrv 为不采用多线程的 DataImportSrvNotUseThread，此类的功能，是对 reader 中的数据，逐行取出，每5000行为一批次，调用数据保存代码。内存占用最多为5000行数据，不会导致内存溢出。

分批时，记录当前批次的数据中，在原始数据文件中的起始行号、结束行号、当前第几批。

DataImportSrvNotUseThread 代码如下：

public class DataImportSrvNotUseThread extends DataImportSrvBase { @Override public void saveData(BufferedReader br, MutableObject<Long> iFromLine, MutableObject<Long> iToLine,   int iBatchCommitSize, String fileName) throws IOException, SQLException {  String strLine = null;  // DataLineParseSrv dataSrv = new DataLineParseSrv();  LinkedList<LineString> batchLineDataBufferList = new LinkedList<LineString>();  long iBatchNum = 0;  long iLineNumOfFile = 0;  while ((strLine = br.readLine()) != null) {   iLineNumOfFile++;   if (iFromLine != null && iFromLine.getValue() > iLineNumOfFile) {    continue;   }   if (iToLine != null && iToLine.getValue() < iLineNumOfFile) {    break;   }   if (StringUtils.isEmpty(strLine)) {    continue;   }   // LineData data = dataSrv.parse(line);   LineString lineData = new LineString();   lineData.strLine = strLine;   lineData.lineNumAtFile = iLineNumOfFile;   batchLineDataBufferList.add(lineData);   if (batchLineDataBufferList.size() >= iBatchCommitSize) {    iBatchNum++;    long iLineNumBeginOfBatch = batchLineDataBufferList.getFirst().lineNumAtFile;    long iLineNumEndOfBatch = lineData.lineNumAtFile;    new DatabaseSrv().saveBatchDataInTrasaction(iBatchNum, iLineNumBeginOfBatch, iLineNumEndOfBatch,      batchLineDataBufferList);    batchLineDataBufferList = new LinkedList<LineString>();   }  }  if (batchLineDataBufferList.size() > 0) {   iBatchNum++;   long iLineNumBeginOfBatch = batchLineDataBufferList.getFirst().lineNumAtFile;   long iLineNumEndOfBatch = batchLineDataBufferList.getLast().lineNumAtFile;   new DatabaseSrv().saveBatchDataInTrasaction(iBatchNum, iLineNumBeginOfBatch, iLineNumEndOfBatch,     batchLineDataBufferList);   batchLineDataBufferList = new LinkedList<LineString>();  } }}

 

最后，DatabaseSrv 类的 saveBatchDataInTrasaction 函数，保存一批数据，使用一个数据库连接、一个 transaction. 此函数内部，使用 PreparedStatement 的executeBatch。

此处使用了数据库连接池。有的数据库，初次建立连接，用时很长，而使用数据库连接池，相比未使用数据库连接池，可大幅提高性能。

DatabaseSrv 代码如下:

public class DatabaseSrv { static BasicDataSource g_ds = null; public void saveBatchDataInTrasaction(long iBatchNum, long iLineNumBeginOfBatch, long iLineNumEndOfBatch,   List<LineString> batchLineDataBufferList) throws SQLException {  Logger log = LoggerFactory.getLogger(DatabaseSrv.class);  log.info("saveBatchDataInTrasaction begin,iBatchNum:" + iBatchNum);  try {   DataLineParseSrv dataSrv = new DataLineParseSrv();   BasicDataSource ds = getDataSource();   try (Connection con = ds.getConnection()) {    con.setAutoCommit(false);    con.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);    String sql = "insert into tt_test(col_a,col_b,col_c,col_d,col_e,col_f,col_g,col_h,col_i,col_j,col_k,col_l) values(?,?,?,?,?,?,?,?,?,?,?,?);";    try (PreparedStatement ps = con.prepareStatement(sql)) {     for (LineString d : batchLineDataBufferList) {      LineParsedData parsedData = dataSrv.parse(d);      long iLineNum = d.lineNumAtFile;// 也可以在表中，先增加一列，保存数据行号。以便检查哪些行成功导入了。      ps.setString(1, parsedData.a); // 1 is the first ? (1 based counting)      ps.setString(2, parsedData.b);      ps.setString(3, parsedData.c);      ps.setString(4, parsedData.d);      ps.setString(5, parsedData.e);      ps.setString(6, parsedData.f);      ps.setString(7, parsedData.g);      ps.setString(8, parsedData.h);      ps.setString(9, parsedData.i);      ps.setString(10, parsedData.j);      ps.setString(11, parsedData.k);      ps.setString(12, parsedData.l);      ps.addBatch();     }     ps.executeBatch();     con.commit();     // statement.clearBatch(); //If you want to add more,    } catch (Exception err) {     log.error(err.getMessage(), err);     con.rollback();    }   }   log.info("批量 commit,批次号:" + iBatchNum + ", 原数据文件行[" + iLineNumBeginOfBatch + "-" + iLineNumEndOfBatch     + "], 提交成功.");  } catch (Exception err) {   log.error(err.getMessage(), err);   log.info("批量 commit,批次号:" + iBatchNum + ", 原数据文件行[" + iLineNumBeginOfBatch + "-" + iLineNumEndOfBatch     + "], 提交失败.");  } } public static BasicDataSource getDataSource() {  if (g_ds != null) {   return g_ds;  } else {   BasicDataSource ds = new BasicDataSource();   ds.setDriverClassName("org.postgresql.Driver");   ds.setTestOnBorrow(true);   ds.setUrl("jdbc:postgresql://192.168.1.50:5432/zg_prt_uld");   ds.setValidationQuery("select 1 as a;");   ds.setUsername("zg_prt_uld_db_user");   ds.setPassword("zg_Hello~1234!");   ds.setInitialSize(1);   ds.setMaxActive(30);   g_ds = ds;   return g_ds;  } }}

 

还有一些重要性较低的代码，此处未贴出。如需要，也可提供。

经初步测试，以上代码，未使用多线程，导入 2万行数据，运行三次，用时分别为 5.696秒, 4.968 秒, 5.04 秒。

按第一次运行的速度(3511行/秒)，导入 2 亿行数据，顺利的话，完成导入所用时间为 15.8小时。即使加上异常数据分析、特殊处理的操作，也能很好完成该博主的工作任务(1周之内完成数据导入)。

 

当然，此处代码，仍有性能优化的余地。

 

以上性能测试，使用的是 Postgres 数据库，本地无线局域网连接。

原文转载：http://www.shaoqun.com/a/504100.html

跨境电商：https://www.ikjzd.com/

扬帆计划：https://www.ikjzd.com/w/1581

upc：https://www.ikjzd.com/w/111

最近在cnblogs网站上，看其他人博客，谈及一个包含很多行(一亿)的大文件，一周之内，将其数据导入到数据库表。我谈到可以使用"使用数据库事务，分批commit到数据库，每批次有5000行"的方法，提高数据导入速度，两天应该就可以了。好像博主及下方评论者，不太理解，这个"分批commit"。特写此博客，介绍一下使用JDBC的executeBatch做分批commit，以提高大批量数据的导入速度。最
贝贝官网：贝贝官网
bap：bap
口述：女友趁我出差 让初恋看光她的身子：口述：女友趁我出差 让初恋看光她的身子
提供大幅折扣也有错？亚马逊和Flipkart遭调查！：提供大幅折扣也有错？亚马逊和Flipkart遭调查！
慈禧曾认为"月饼"说法不雅 改名为"月菜糕"：慈禧曾认为"月饼"说法不雅 改名为"月菜糕"