java 大文件 處理

我最近不得不處理一組包含歷史逐筆交易的外匯市場數據的文件，并很快意識到使用傳統的InputStream都無法將它們讀取到內存中，因為每個文件的大小都超過4 GB。 Emacs甚至無法打開它們。

在這種特殊情況下，我可以編寫一個簡單的bash腳本，將文件分成小塊，然后像往常一樣讀取它們。 但是我不希望這樣，因為二進制格式會使這種方法無效。

因此，正確處理此問題的方法是使用內存映射文件逐步處理數據區域。 內存映射文件的優點在于它們不消耗虛擬內存或分頁空間，因為它們由磁盤上的文件數據支持。

Okey，讓我們看一下這些文件并提取一些數據。 似乎它們包含帶有逗號分隔字段的ASCII文本行。

格式： [currency-pair],[timestamp],[bid-price],[ask-price]

例如： EUR/USD,20120102 00:01:30.420,1.29451,1.2949

公平地說，我可以為該格式編寫程序。 但是讀取和解析文件是正交的概念。 因此，讓我們退后一步，考慮一下可以在將來遇到類似問題時可以重用的通用設計。

問題歸結為對一組以無限長字節數組編碼的條目進行增量解碼，而不會耗盡內存。 示例格式以逗號/行分隔文本編碼的事實與一般解決方案無關，因此很明顯，需要解碼器接口才能處理不同的格式。

同樣，在處理完整個文件之前，無法解析每個條目并將其保留在內存中，因此我們需要一種方法來逐步移交可以在其他位置（磁盤或網絡）寫入的條目塊，然后再進行垃圾回收。 迭代器是處理此要求的很好的抽象方法，因為它們的行為就像游標一樣，這正是重點。 每次迭代都會轉發文件指針，然后讓我們對數據進行處理。

所以首先是Decoder接口。 這個想法是從MappedByteBuffer增量解碼對象，或者如果緩沖區中沒有對象，則返回null。

public interface Decoder<T> {public T decode(ByteBuffer buffer); }

然后是實現Iterable的FileReader 。 每次迭代將處理下一個4096字節的數據，并使用Decoder將其Decoder為對象列表。 請注意， FileReader接受文件列表，這很不錯，因為它允許遍歷數據而不必擔心跨文件聚合。 順便說一下，對于較大的文件，4096個字節的塊可能會有點小。

public class FileReader implements Iterable<List<T>> {private static final long CHUNK_SIZE = 4096;private final Decoder<T> decoder;private Iterator<File> files;private FileReader(Decoder<T> decoder, File... files) {this(decoder, Arrays.asList(files));}private FileReader(Decoder<T> decoder, List<File> files) {this.files = files.iterator();this.decoder = decoder;}public static <T> FileReader<T> create(Decoder<T> decoder, List<File> files) {return new FileReader<T>(decoder, files);}public static <T> FileReader<T> create(Decoder<T> decoder, File... files) {return new FileReader<T>(decoder, files);}@Overridepublic Iterator<List<T>> iterator() {return new Iterator<List<T>>() {private List<T> entries;private long chunkPos = 0;private MappedByteBuffer buffer;private FileChannel channel;@Overridepublic boolean hasNext() {if (buffer == null || !buffer.hasRemaining()) {buffer = nextBuffer(chunkPos);if (buffer == null) {return false;}}T result = null;while ((result = decoder.decode(buffer)) != null) {if (entries == null) {entries = new ArrayList<T>();}entries.add(result);}// set next MappedByteBuffer chunkchunkPos += buffer.position();buffer = null;if (entries != null) {return true;} else {Closeables.closeQuietly(channel);return false;}}private MappedByteBuffer nextBuffer(long position) {try {if (channel == null || channel.size() == position) {if (channel != null) {Closeables.closeQuietly(channel);channel = null;}if (files.hasNext()) {File file = files.next();channel = new RandomAccessFile(file, "r").getChannel();chunkPos = 0;position = 0;} else {return null;}}long chunkSize = CHUNK_SIZE;if (channel.size() - position < chunkSize) {chunkSize = channel.size() - position;}return channel.map(FileChannel.MapMode.READ_ONLY, chunkPos, chunkSize);} catch (IOException e) {Closeables.closeQuietly(channel);throw new RuntimeException(e);}}@Overridepublic List<T> next() {List<T> res = entries;entries = null;return res;}@Overridepublic void remove() {throw new UnsupportedOperationException();}};} }

下一個任務是編寫一個Decoder ，我決定為任何逗號分隔的文本文件格式實現一個通用的TextRowDecoder ，接受每行的字段數和一個字段定界符，并返回一個字節數組數組。 然后， TextRowDecoder可以由格式特定的解碼器重用，這些解碼器可以處理不同的字符集。

public class TextRowDecoder implements Decoder<byte[][]> {private static final byte LF = 10;private final int numFields;private final byte delimiter;public TextRowDecoder(int numFields, byte delimiter) {this.numFields = numFields;this.delimiter = delimiter;}@Overridepublic byte[][] decode(ByteBuffer buffer) {int lineStartPos = buffer.position();int limit = buffer.limit();while (buffer.hasRemaining()) {byte b = buffer.get();if (b == LF) { // reached line feed so parse lineint lineEndPos = buffer.position();// set positions for one row duplicationif (buffer.limit() < lineEndPos + 1) {buffer.position(lineStartPos).limit(lineEndPos);} else {buffer.position(lineStartPos).limit(lineEndPos + 1);}byte[][] entry = parseRow(buffer.duplicate());if (entry != null) {// reset main bufferbuffer.position(lineEndPos);buffer.limit(limit);// set start after LFlineStartPos = lineEndPos;}return entry;}}buffer.position(lineStartPos);return null;}public byte[][] parseRow(ByteBuffer buffer) {int fieldStartPos = buffer.position();int fieldEndPos = 0;int fieldNumber = 0;byte[][] fields = new byte[numFields][];while (buffer.hasRemaining()) {byte b = buffer.get();if (b == delimiter || b == LF) {fieldEndPos = buffer.position();// save limitint limit = buffer.limit();// set positions for one row duplicationbuffer.position(fieldStartPos).limit(fieldEndPos);fields[fieldNumber] = parseField(buffer.duplicate(), fieldNumber, fieldEndPos - fieldStartPos - 1);fieldNumber++;// reset main bufferbuffer.position(fieldEndPos);buffer.limit(limit);// set start after LFfieldStartPos = fieldEndPos;}if (fieldNumber == numFields) {return fields;}}return null;}private byte[] parseField(ByteBuffer buffer, int pos, int length) {byte[] field = new byte[length];for (int i = 0; i < field.length; i++) {field[i] = buffer.get();}return field;} }

這就是文件的處理方式。 每個列表包含從單個緩沖區解碼的元素，每個元素都是由TextRowDecoder指定的字節數組的數組。

TextRowDecoder decoder = new TextRowDecoder(4, comma); FileReader<byte[][]> reader = FileReader.create(decoder, file.listFiles()); for (List<byte[][]> chunk : reader) {// do something with each chunk }

我們可以在這里停下來，但還有其他要求。 每行都包含一個時間戳記，并且必須按時間段而不是按天或按小時對緩沖區進行分組。 我仍然想遍歷每個批處理，因此立即的React是為FileReader創建一個Iterable包裝器，以實現此行為。 另外一個細節是，每個元素都必須通過實現PeriodEntries Timestamped接口（此處未顯示）為PeriodEntries提供其時間戳。

public class PeriodEntries<T extends Timestamped> implements Iterable<List<T>> {private final Iterator<List<T extends Timestamped>> entriesIt;private final long interval;private PeriodEntries(Iterable<List<T>> entriesIt, long interval) {this.entriesIt = entriesIt.iterator();this.interval = interval;}public static <T extends Timestamped> PeriodEntries<T> create(Iterable<List<T>> entriesIt, long interval) {return new PeriodEntries<T>(entriesIt, interval);}@Overridepublic Iterator<List<T extends Timestamped>> iterator() {return new Iterator<List<T>>() {private Queue<List<T>> queue = new LinkedList<List<T>>();private long previous;private Iterator<T> entryIt;@Overridepublic boolean hasNext() {if (!advanceEntries()) {return false;}T entry = entryIt.next();long time = normalizeInterval(entry);if (previous == 0) {previous = time;}if (queue.peek() == null) {List<T> group = new ArrayList<T>();queue.add(group);}while (previous == time) {queue.peek().add(entry);if (!advanceEntries()) {break;}entry = entryIt.next();time = normalizeInterval(entry);}previous = time;List<T> result = queue.peek();if (result == null || result.isEmpty()) {return false;}return true;}private boolean advanceEntries() {// if there are no rows leftif (entryIt == null || !entryIt.hasNext()) {// try get more rows if possibleif (entriesIt.hasNext()) {entryIt = entriesIt.next().iterator();return true;} else {// no more rowsreturn false;}}return true;}private long normalizeInterval(Timestamped entry) {long time = entry.getTime();int utcOffset = TimeZone.getDefault().getOffset(time);long utcTime = time + utcOffset;long elapsed = utcTime % interval;return time - elapsed;}@Overridepublic List<T> next() {return queue.poll();}@Overridepublic void remove() {throw new UnsupportedOperationException();}};} }

引入此功能后，最終處理代碼并沒有太大變化，只是一個干凈緊湊的for循環而不必關心跨文件，緩沖區和句點對元素進行分組。 PeriodEntries也足夠靈活，可以管理間隔上的任何長度。

TrueFxDecoder decoder = new TrueFxDecoder(); FileReader<TrueFxData> reader = FileReader.create(decoder, file.listFiles()); long periodLength = TimeUnit.DAYS.toMillis(1); PeriodEntries<TrueFxData> periods = PeriodEntries.create(reader, periodLength);for (List<TrueFxData> entries : periods) {// data for each dayfor (TrueFxData entry : entries) {// process each entry} }

如您所知，不可能用集合來解決這個問題。 選擇迭代器是一項關鍵的設計決策，它能夠解析TB級的數據而不會占用太多的堆空間。

參考： Deephacks博客上的JCG合作伙伴 Kristoffer Sjogren 使用Java處理大型文件 。

翻譯自: https://www.javacodegeeks.com/2013/01/processing-huge-files-with-java.html

java 大文件 處理

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的java 大文件 处理_用Java处理大文件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。