最近在 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], 提交成功.
如果 5001-10000 行处理失败,则下次运行时,更改启动参数 from_line=5001, to_line=10000, batch_commit_size 调小为 100,再次运行。
将出错行号,逐步定位到更小的批次里。
主控程序 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 iFromLine = null;
if (StringUtils.isNotEmpty(strFromLine)) {
iFromLine = new MutableObject();
iFromLine.setValue(Long.parseLong(strFromLine));
}
MutableObject iToLine = null;
if (StringUtils.isNotEmpty(strToLine)) {
iToLine = new MutableObject();
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,此 class 的功能,是将 reader 中的数据,逐行取出,每5000行为一批次,调用数据保存代码。
内存占用最多为5000行数据,不会导致内存溢出。
分批时,记录当前批次的数据中,在原始数据文件中的起始行号、结束行号、当前第几批。
DataImportSrvNotUseThread 代码如下:
public class DataImportSrvNotUseThread extends DataImportSrvBase {
@Override
public void saveData(BufferedReader br, MutableObject iFromLine, MutableObject iToLine,
int iBatchCommitSize, String fileName) throws IOException, SQLException {
String strLine = null;
// DataLineParseSrv dataSrv = new DataLineParseSrv();
LinkedList batchLineDataBufferList = new LinkedList();
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();
}
}
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();
}
}
}
最后,DatabaseSrv 类的 saveBatchDataInTrasaction 函数,保存一批数据,使用一个数据库连接、一个 transaction. 此函数内部,使用 PreparedStatement 的executeBatch。
此处使用了数据库连接池。
有的数据库,初次建立连接,用时很长,而使用数据库连接池,相比未使用数据库连接池,可大幅提高性能。
DatabaseSrv 代码如下:
public class DatabaseSrv {
static BasicDataSource g_ds = null;
public void saveBatchDataInTrasaction(long iBatchNum, long iLineNumBeginOfBatch, long iLineNumEndOfBatch,
List batchLineDataBufferList) throws SQLException {
Logger log = LoggerFactory.getLogger(DatabaseSrv.class);
log.info("saveBatchDataInTrasaction begin,iBatchNum:" + iBatchNum);
boolean bCommitSuccess = false;
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();
// 标记为成功
bCommitSuccess = true;
// statement.clearBatch(); //If you want to add more,
} catch (Exception err) {
log.error(err.getMessage(), err);
con.rollback();
}
}
} catch (Exception err) {
log.error(err.getMessage(), err);
}
if (bCommitSuccess) {
log.info("批量 commit,批次号:" + iBatchNum + ", 原数据文件行[" + iLineNumBeginOfBatch + "-" + iLineNumEndOfBatch
+ "], 提交成功.");
} else {
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("xxxx");
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 数据库,本地无线局域网连接。