高效读取大数据文本文件（上亿行数据）
程序员文章站
2022-06-13 23:09:52
...
一.前言
本文是对大数据文本文件读取（按行读取）的优化，目前常规的方案（限于JDK）有三种，第一种LineNumberReader，第二种RandomAccessFile，第三种是内存映射文件（详见http://sgq0085.iteye.com/blog/1318622）在RandomAccessFile基础上调用getChannel().map(...)。
1.LineNumberReader
按行读取，只能从第一行向后遍历，到需要读取的行时开始读入，直到完成；在我的测试用例中，读取1000W行数据每次5万行，用时93秒，效率实测比RandomAccessFile要高，但读取一亿跳数据时效率太低了（因为每次都要从头遍历），因为测试时超过1个小时，放弃测试；
2.RandomAccessFile
实际不适用于这种大数据读取，RandomAccessFile是为了磁盘文件的随机访问，所以效率很低，1000w行测试时用时140秒，一亿行数据测试用时1438秒但由于可以通过getFilePointer方法记录位置，并通过seek方法指定读取位置，所以从理论上比较适用这种大数据按行读取的场景；
RandomAccessFile只能按照8859_1这种方法读取，所以需要对内容重新编码，方法如下
new String(pin.getBytes("8859_1"), "")
3.内存映射文件
由于每行数据大小不同，内存映射文件在这种情况下不适用，其他情况请参考我的博客（详见http://sgq0085.iteye.com/blog/1318622）
二.解决方案
如果在RandomAccessFile基础上，整合内部缓冲区，效率会有提高，测试过程中1000w行数据用时1秒，1亿行数据用时103（比1438秒快了13倍左右）
BufferedRandomAccessFile
网上已经有实现，代码如下：
package com.gqshao.file.io;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Arrays;
public class BufferedRandomAccessFile extends RandomAccessFile {
static final int LogBuffSz_ = 16; // 64K buffer
public static final int BuffSz_ = (1 << LogBuffSz_);
static final long BuffMask_ = ~(((long) BuffSz_) - 1L);
private String path_;
/*
* This implementation is based on the buffer implementation in Modula-3's
* "Rd", "Wr", "RdClass", and "WrClass" interfaces.
*/
private boolean dirty_; // true iff unflushed bytes exist
private boolean syncNeeded_; // dirty_ can be cleared by e.g. seek, so track sync separately
private long curr_; // current position in file
private long lo_, hi_; // bounds on characters in "buff"
private byte[] buff_; // local buffer
private long maxHi_; // this.lo + this.buff.length
private boolean hitEOF_; // buffer contains last file block?
private long diskPos_; // disk position
/*
* To describe the above fields, we introduce the following abstractions for
* the file "f":
*
* len(f) the length of the file curr(f) the current position in the file
* c(f) the abstract contents of the file disk(f) the contents of f's
* backing disk file closed(f) true iff the file is closed
*
* "curr(f)" is an index in the closed interval [0, len(f)]. "c(f)" is a
* character sequence of length "len(f)". "c(f)" and "disk(f)" may differ if
* "c(f)" contains unflushed writes not reflected in "disk(f)". The flush
* operation has the effect of making "disk(f)" identical to "c(f)".
*
* A file is said to be *valid* if the following conditions hold:
*
* V1. The "closed" and "curr" fields are correct:
*
* f.closed == closed(f) f.curr == curr(f)
*
* V2. The current position is either contained in the buffer, or just past
* the buffer:
*
* f.lo <= f.curr <= f.hi
*
* V3. Any (possibly) unflushed characters are stored in "f.buff":
*
* (forall i in [f.lo, f.curr): c(f)[i] == f.buff[i - f.lo])
*
* V4. For all characters not covered by V3, c(f) and disk(f) agree:
*
* (forall i in [f.lo, len(f)): i not in [f.lo, f.curr) => c(f)[i] ==
* disk(f)[i])
*
* V5. "f.dirty" is true iff the buffer contains bytes that should be
* flushed to the file; by V3 and V4, only part of the buffer can be dirty.
*
* f.dirty == (exists i in [f.lo, f.curr): c(f)[i] != f.buff[i - f.lo])
*
* V6. this.maxHi == this.lo + this.buff.length
*
* Note that "f.buff" can be "null" in a valid file, since the range of
* characters in V3 is empty when "f.lo == f.curr".
*
* A file is said to be *ready* if the buffer contains the current position,
* i.e., when:
*
* R1. !f.closed && f.buff != null && f.lo <= f.curr && f.curr < f.hi
*
* When a file is ready, reading or writing a single byte can be performed
* by reading or writing the in-memory buffer without performing a disk
* operation.
*/
/**
* Open a new <code>BufferedRandomAccessFile</code> on <code>file</code>
* in mode <code>mode</code>, which should be "r" for reading only, or
* "rw" for reading and writing.
*/
public BufferedRandomAccessFile(File file, String mode) throws IOException {
this(file, mode, 0);
}
public BufferedRandomAccessFile(File file, String mode, int size) throws IOException {
super(file, mode);
path_ = file.getAbsolutePath();
this.init(size);
}
/**
* Open a new <code>BufferedRandomAccessFile</code> on the file named
* <code>name</code> in mode <code>mode</code>, which should be "r" for
* reading only, or "rw" for reading and writing.
*/
public BufferedRandomAccessFile(String name, String mode) throws IOException {
this(name, mode, 0);
}
public BufferedRandomAccessFile(String name, String mode, int size) throws FileNotFoundException {
super(name, mode);
path_ = name;
this.init(size);
}
private void init(int size) {
this.dirty_ = false;
this.lo_ = this.curr_ = this.hi_ = 0;
this.buff_ = (size > BuffSz_) ? new byte[size] : new byte[BuffSz_];
this.maxHi_ = (long) BuffSz_;
this.hitEOF_ = false;
this.diskPos_ = 0L;
}
public String getPath() {
return path_;
}
public void sync() throws IOException {
if (syncNeeded_) {
flush();
getChannel().force(true);
syncNeeded_ = false;
}
}
// public boolean isEOF() throws IOException
// {
// assert getFilePointer() <= length();
// return getFilePointer() == length();
// }
public void close() throws IOException {
this.flush();
this.buff_ = null;
super.close();
}
/**
* Flush any bytes in the file's buffer that have not yet been written to
* disk. If the file was created read-only, this method is a no-op.
*/
public void flush() throws IOException {
this.flushBuffer();
}
/* Flush any dirty bytes in the buffer to disk. */
private void flushBuffer() throws IOException {
if (this.dirty_) {
if (this.diskPos_ != this.lo_)
super.seek(this.lo_);
int len = (int) (this.curr_ - this.lo_);
super.write(this.buff_, 0, len);
this.diskPos_ = this.curr_;
this.dirty_ = false;
}
}
/*
* Read at most "this.buff.length" bytes into "this.buff", returning the
* number of bytes read. If the return result is less than
* "this.buff.length", then EOF was read.
*/
private int fillBuffer() throws IOException {
int cnt = 0;
int rem = this.buff_.length;
while (rem > 0) {
int n = super.read(this.buff_, cnt, rem);
if (n < 0)
break;
cnt += n;
rem -= n;
}
if ((cnt < 0) && (this.hitEOF_ = (cnt < this.buff_.length))) {
// make sure buffer that wasn't read is initialized with -1
Arrays.fill(this.buff_, cnt, this.buff_.length, (byte) 0xff);
}
this.diskPos_ += cnt;
return cnt;
}
/*
* This method positions <code>this.curr</code> at position <code>pos</code>.
* If <code>pos</code> does not fall in the current buffer, it flushes the
* current buffer and loads the correct one.<p>
*
* On exit from this routine <code>this.curr == this.hi</code> iff <code>pos</code>
* is at or past the end-of-file, which can only happen if the file was
* opened in read-only mode.
*/
public void seek(long pos) throws IOException {
if (pos >= this.hi_ || pos < this.lo_) {
// seeking outside of current buffer -- flush and read
this.flushBuffer();
this.lo_ = pos & BuffMask_; // start at BuffSz boundary
this.maxHi_ = this.lo_ + (long) this.buff_.length;
if (this.diskPos_ != this.lo_) {
super.seek(this.lo_);
this.diskPos_ = this.lo_;
}
int n = this.fillBuffer();
this.hi_ = this.lo_ + (long) n;
} else {
// seeking inside current buffer -- no read required
if (pos < this.curr_) {
// if seeking backwards, we must flush to maintain V4
this.flushBuffer();
}
}
this.curr_ = pos;
}
public long getFilePointer() {
return this.curr_;
}
public long length() throws IOException {
// max accounts for the case where we have written past the old file length, but not yet flushed our buffer
return Math.max(this.curr_, super.length());
}
public int read() throws IOException {
if (this.curr_ >= this.hi_) {
// test for EOF
// if (this.hi < this.maxHi) return -1;
if (this.hitEOF_)
return -1;
// slow path -- read another buffer
this.seek(this.curr_);
if (this.curr_ == this.hi_)
return -1;
}
byte res = this.buff_[(int) (this.curr_ - this.lo_)];
this.curr_++;
return ((int) res) & 0xFF; // convert byte -> int
}
public int read(byte[] b) throws IOException {
return this.read(b, 0, b.length);
}
public int read(byte[] b, int off, int len) throws IOException {
if (this.curr_ >= this.hi_) {
// test for EOF
// if (this.hi < this.maxHi) return -1;
if (this.hitEOF_)
return -1;
// slow path -- read another buffer
this.seek(this.curr_);
if (this.curr_ == this.hi_)
return -1;
}
len = Math.min(len, (int) (this.hi_ - this.curr_));
int buffOff = (int) (this.curr_ - this.lo_);
System.arraycopy(this.buff_, buffOff, b, off, len);
this.curr_ += len;
return len;
}
public void write(int b) throws IOException {
if (this.curr_ >= this.hi_) {
if (this.hitEOF_ && this.hi_ < this.maxHi_) {
// at EOF -- bump "hi"
this.hi_++;
} else {
// slow path -- write current buffer; read next one
this.seek(this.curr_);
if (this.curr_ == this.hi_) {
// appending to EOF -- bump "hi"
this.hi_++;
}
}
}
this.buff_[(int) (this.curr_ - this.lo_)] = (byte) b;
this.curr_++;
this.dirty_ = true;
syncNeeded_ = true;
}
public void write(byte[] b) throws IOException {
this.write(b, 0, b.length);
}
public void write(byte[] b, int off, int len) throws IOException {
while (len > 0) {
int n = this.writeAtMost(b, off, len);
off += n;
len -= n;
this.dirty_ = true;
syncNeeded_ = true;
}
}
/*
* Write at most "len" bytes to "b" starting at position "off", and return
* the number of bytes written.
*/
private int writeAtMost(byte[] b, int off, int len) throws IOException {
if (this.curr_ >= this.hi_) {
if (this.hitEOF_ && this.hi_ < this.maxHi_) {
// at EOF -- bump "hi"
this.hi_ = this.maxHi_;
} else {
// slow path -- write current buffer; read next one
this.seek(this.curr_);
if (this.curr_ == this.hi_) {
// appending to EOF -- bump "hi"
this.hi_ = this.maxHi_;
}
}
}
len = Math.min(len, (int) (this.hi_ - this.curr_));
int buffOff = (int) (this.curr_ - this.lo_);
System.arraycopy(b, off, this.buff_, buffOff, len);
this.curr_ += len;
return len;
}
}
三.测试
1.FileUtil
用于封装三种方案（LineNumberReader、RandomAccessFile、BufferedRandomAccessFile）的文件读取
package com.gqshao.file.util;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.gqshao.file.io.BufferedRandomAccessFile;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import java.io.*;
import java.util.List;
import java.util.Map;
public class FileUtil {
/**
* 通过BufferedRandomAccessFile读取文件,推荐
*
* @param file 源文件
* @param encoding 文件编码
* @param pos 偏移量
* @param num 读取量
* @return pins文件内容，pos当前偏移量
*/
public static Map<String, Object> BufferedRandomAccessFileReadLine(File file, String encoding, long pos, int num) {
Map<String, Object> res = Maps.newHashMap();
List<String> pins = Lists.newArrayList();
res.put("pins", pins);
BufferedRandomAccessFile reader = null;
try {
reader = new BufferedRandomAccessFile(file, "r");
reader.seek(pos);
for (int i = 0; i < num; i++) {
String pin = reader.readLine();
if (StringUtils.isBlank(pin)) {
break;
}
pins.add(new String(pin.getBytes("8859_1"), encoding));
}
res.put("pos", reader.getFilePointer());
} catch (Exception e) {
e.printStackTrace();
} finally {
IOUtils.closeQuietly(reader);
}
return res;
}
/**
* 通过RandomAccessFile读取文件，能出来大数据文件，效率低
*
* @param file 源文件
* @param encoding 文件编码
* @param pos 偏移量
* @param num 读取量
* @return pins文件内容，pos当前偏移量
*/
public static Map<String, Object> readLine(File file, String encoding, long pos, int num) {
Map<String, Object> res = Maps.newHashMap();
List<String> pins = Lists.newArrayList();
res.put("pins", pins);
RandomAccessFile reader = null;
try {
reader = new RandomAccessFile(file, "r");
reader.seek(pos);
for (int i = 0; i < num; i++) {
String pin = reader.readLine();
if (StringUtils.isBlank(pin)) {
break;
}
pins.add(new String(pin.getBytes("8859_1"), encoding));
}
res.put("pos", reader.getFilePointer());
} catch (Exception e) {
e.printStackTrace();
} finally {
IOUtils.closeQuietly(reader);
}
return res;
}
/**
* 使用LineNumberReader读取文件，1000w行比RandomAccessFile效率高，无法处理1亿条数据
*
* @param file 源文件
* @param encoding 文件编码
* @param index 开始位置
* @param num 读取量
* @return pins文件内容
*/
public static List<String> readLine(File file, String encoding, int index, int num) {
List<String> pins = Lists.newArrayList();
LineNumberReader reader = null;
try {
reader = new LineNumberReader(new InputStreamReader(new FileInputStream(file), encoding));
int lines = 0;
while (true) {
String pin = reader.readLine();
if (StringUtils.isBlank(pin)) {
break;
}
if (lines >= index) {
if (StringUtils.isNotBlank(pin)) {
pins.add(pin);
}
}
if (num == pins.size()) {
break;
}
lines++;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
IOUtils.closeQuietly(reader);
}
return pins;
}
}
2.RandomAccessFileTest
测试方法，涉及到的randomFile只是一个掺杂中文的文本文件，可以自己随便写一个
package com.gqshao.file;
import com.gqshao.file.util.FileUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.io.IOUtils;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.util.List;
import java.util.Map;
public class RandomAccessFileTest {
private static final Logger logger = LoggerFactory.getLogger(RandomAccessFileTest.class);
private static final String ENCODING = "UTF-8";
private static final int NUM = 50000;
private static File file = new File(ClassLoader.getSystemResource("").getPath() + File.separator + "test.txt");
private static File randomFile = new File(ClassLoader.getSystemResource("").getPath() + File.separator + "RandomFile.txt");
/**
* 生成1000w随机文本文件
*/
@Test
public void makePin() {
String prefix = "_$#";
OutputStreamWriter out = null;
try {
out = new OutputStreamWriter(new FileOutputStream(file, true), ENCODING);
// 在1500w里随机1000w数据
for (int j = 0; j < 100000000; j++) {
out.write(prefix + (int) (130000000 * Math.random()) + "\n");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
IOUtils.closeQuietly(out);
}
logger.info(file.getAbsolutePath());
}
/**
* 测试RandomAccessFile读取文件
*/
@Test
public void testRandomAccessRead() {
long start = System.currentTimeMillis();
//
logger.info(String.valueOf(file.exists()));
long pos = 0L;
while (true) {
Map<String, Object> res = FileUtil.readLine(file, ENCODING, pos, NUM);
// 如果返回结果为空结束循环
if (MapUtils.isEmpty(res)) {
break;
}
Object po = res.get("pins");
List<String> pins = (List<String>) res.get("pins");
if (CollectionUtils.isNotEmpty(pins)) {
// logger.info(Arrays.toString(pins.toArray()));
if (pins.size() < NUM) {
break;
}
} else {
break;
}
pos = (Long) res.get("pos");
}
logger.info(((System.currentTimeMillis() - start) / 1000) + "");
}
/**
* 测试RandomAccessFile读取文件
*/
@Test
public void testBufferedRandomAccessRead() {
long start = System.currentTimeMillis();
//
logger.info(String.valueOf(file.exists()));
long pos = 0L;
while (true) {
Map<String, Object> res = FileUtil.BufferedRandomAccessFileReadLine(file, ENCODING, pos, NUM);
// 如果返回结果为空结束循环
if (MapUtils.isEmpty(res)) {
break;
}
List<String> pins = (List<String>) res.get("pins");
if (CollectionUtils.isNotEmpty(pins)) {
// logger.info(Arrays.toString(pins.toArray()));
if (pins.size() < NUM) {
break;
}
} else {
break;
}
pos = (Long) res.get("pos");
}
logger.info(((System.currentTimeMillis() - start) / 1000) + "");
}
/**
* 测试普通读取文件
*/
@Test
public void testCommonRead() {
long start = System.currentTimeMillis();
logger.info(String.valueOf(randomFile.exists()));
int index = 0;
while (true) {
List<String> pins = FileUtil.readLine(file, ENCODING, index, NUM);
if (CollectionUtils.isNotEmpty(pins)) {
// logger.info(Arrays.toString(pins.toArray()));
if (pins.size() < NUM) {
break;
}
} else {
break;
}
index += NUM;
}
logger.info(((System.currentTimeMillis() - start) / 1000) + "");
}
}
上一篇： 如何优雅的使用异常