【Python】容量の大きなファイルを後ろから読み込む方法

この記事を読むとできること

下記のような場合に、メモリと時間を使い過ぎずにファイルを読み込むことができます。

• 必要なデータは最後の方にある
• ファイルの容量が大きい
• 大量のファイルを処理する

はじめに

大学での研究で解析をしている際に必要なデータは最後の方にあるかつ大量のファイルを処理するので、ファイルを後ろから読み込み解析することで効率的に解析したいということがありました。

ファイルの容量が小さい場合は下記のようにすべて読み込んで反対からループすれば良いです。

with open('sample.txt') as f:
    lines = f.readlines()
    
for line in reversed(lines):
    print(l)

しかし、私が解析するデータはファイルの容量が大きく、すべて読み込むことは難しいです。

そこで、容量の大きいファイルをメモリを使い過ぎずに後ろから読み込むコードを作成しました。

本文

コード全体

最初にコード全体を載せておきます。
処理の中身に興味ない人は下記のコードをコピーしてmain()のように使用してください。

import os


class FileReader:
    def __init__(self, file_path, buffer_size=8192):
        self.buffer_size = buffer_size
        self.file = open(file_path, 'rb')
        # ファイルポインタをファイルの末尾に移動
        self.file.seek(0, os.SEEK_END)
        self.file_size = self.file.tell()
        # 最後のブロックの終了位置をファイルサイズに設定
        self.reversed_block_end = self.file_size
        # 最後のブロックの開始位置を計算（ファイル末尾からバッファサイズ分前、またはファイル先頭）
        self.reversed_block_start = max(0, self.reversed_block_end - self.buffer_size)
        self.lines = []

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def close(self):
        self.file.close()

    def reversed_readline(self):
        if not self.lines and self.reversed_block_end - self.reversed_block_start != 0:
            # ファイルポインタを現在のブロックの開始位置に設定
            self.file.seek(self.reversed_block_start)

            block = self.file.read(self.reversed_block_end - self.reversed_block_start)
            lines = block.splitlines(keepends=True)

            # 現在のブロックがファイルの最初のブロックの場合
            if self.reversed_block_start == 0:
                # ブロックの終了位置を開始位置に設定して次の読み取りを防ぐ
                self.reversed_block_end = self.reversed_block_start
            else:
                # 次のブロックの終了位置を更新(現在のブロックの最初の行の長さを加算)
                self.reversed_block_end = max(0, self.reversed_block_start + len(lines[0]))

            self.reversed_block_start = max(0, self.reversed_block_end - self.buffer_size)

            # 末尾のブロックの場合、全行を読み取る
            if self.reversed_block_end == 0:
                self.lines = lines
            else:
                self.lines = lines[1:]

            return self.lines.pop().decode('utf-8')
        elif self.lines:
            return self.lines.pop().decode('utf-8')
        else:
            return None


def main():
    with FileReader('sample.txt') as f:
        while True:
            line = f.reversed_readline()
            if not line:
                break
            print(line)


if __name__ == '__main__':
    main()

解説

ここからは最初に示したコードの処理の中身を解説していきます。

初期化メソッド

まず、ファイルをバイナリモードで開きます。

次にself.file.seek(0, os.SEEK_END)でファイルポインタをファイルの末尾に移動させます。
ファイルポインタは、ファイル内の現在の読み書き位置を示します。

次に、後ろからファイルを読み込むためにファイルの全体のサイズを取得します。

最後に、先頭へ指定したバッファサイズ分移動させて初期化メソッドは終了です。
ここでは万が一、先頭を超えポインタがマイナスにならないようにmax()を使用しています。

reversed_readline()

self.lines=[]が空かつすべてファイルを読み込めていない場合、現在のファイルポインタから前回のファイルポインタまで読み込みます。
(1回目: 現在のファイルポインタは末尾からバッファサイズ分移動した位置、前回のファイルポインタは末尾)

ここで、中途半端な位置(今回は改行区切りにした)まで読み込んでしまった場合は、ファイルポインタの位置を改行の位置に移動させて読み込みます。

そして、self.lines=[]が空になるまで、reversed_readline()が呼ばれるたびにself.linesの後ろからデコードして文章を返します。

このプロセスを先頭まで文章を読み込めるまで繰り返します。

おわりに

容量の大きなファイルを後ろから読み込む方法を紹介しました。
上手く解説できている気がしないですが、誰かの助けになると嬉しいです。