【Python】SQLite で日本語を全文検索するコード例【N-Gram, FTS4/FTS5】

日本語の全文検索ぜんぶんけんさく (full-text search, FTS) を、高速に実行する Python コード例です。

Python の標準モジュール sqlite3 を使用しました。

sqlite3 から、SQLiteエスキューライトの全文検索 (FTSエフティーエス) を使ってみました。

試したのは、FTS4エフティーエスフォーと FTS5エフティーエスファイブの２種類です。

ところで、SQLite の読み方は色々ありました。YouTube では、エスキューライト、エスキューエライト、スィクライト、スィクエライト、などの発音を聞きました。

全文検索の使い方（FTS の使い方）ですが、テキストを N-Gram にして、FTS4 か FTS5 の仮想テーブルに INSERT するだけでした。

（2022年2月5日追記）MeCab の使い方も書きました。

（追記終わり）

自分は、bigramバイグラム (2-Gram) にして INSERT しました。

検索するときは、検索キーワードを N-Gram にして、MATCH "N-Gram" で検索したらヒットしました。

（N-Gram の例）"増収増益" ⇒ "増収収増増益"

あとは、ヒットした N-Gram から、元のテキストを復元して、検索結果として表示しました。

（復元例）"増収収増増益" ⇒ "増収増益"

とても簡単でした。

難しい転置てんちインデックスは、SQLite が自動的に管理してくれました。

インデックスの更新も簡単でした。

テキストは、あとからどんどん追加 (INSERT) していくことができましたし、テキストをインデックスから削除 (DELETE) することもできました。

検索速度も、通常のインデックス無しの検索と比べて、圧倒的に高速でした。

自分は、XBRL からテキストを取得して、自分専用の完全ローカル検索エンジンを作ってみたのですが、良かったです。

30GB 以上あった開示情報の XBRL から、狙った文章をサクッと列挙れっきょすることができました。

FTS で MATCH 検索に要した時間は、SSD 上でも HDD 上でも、たいてい１秒未満でした。

時間がかかった場合でも、数秒から数分でした。

特別な工夫無しで、この速さでした。

SQLite の全文検索は、手軽で高速で実用的で、とても良かったです。

もちろん、HTML からでも、PDF からでも、テキストさえ取得できれば、検索対象に含めることができました。

SQLite の FTS は、『人間が使う検索エンジン』としても役に立ちましたし、『テキスト分析システムが検索するための検索エンジン』としても、力を発揮してくれそうでした。

Python マニュアルの場所
SQLite マニュアルの場所
コード例
実行結果
SQLite DB を最適化する方法
1. FTS 仮想テーブルを最適化するコード例
2. SQLite DB を VACUUM するコード例
XBRL テキスト検索システムのスクリーンショット
SQLite の FTS4 に格納できるテキストの限界

Python マニュアルの場所

コード例で使用した Python 機能のマニュアルです。

SQLite マニュアルの場所

コード例で使用した SQLite 機能のマニュアルです。

コード例

全文検索用の SQLite DB を作る Python コード例です。

FTS5 で仮想テーブルを作って、テキストを追加してから、試しに検索してみました。

main()

メイン関数です。

テキストデータを用意して、データベースを開いて、カーソルを取得しました。

SQLite の全文検索は、FTS5 を使うこともできましたし、FTS4 を使うこともできました。

CREATE VIRTUAL TABLE の書き方が、少しちがっていただけでした。

FTS5 は、SQLite 組み込み補助関数の bm25() を、手軽に使えたところがよかったです。

コード例では割愛かつあいしましたが、検索結果を、関連度順かんれんどじゅんに並べかえることができました。

(SQLite) The bm25() function – SELECT * FROM fts WHERE fts MATCH ? ORDER BY bm25(fts)

"""
make_fts_table.py
日本語の全文検索 (full-text search, FTS) をする
Python コード例です。Python の SQLite を使用しました。
SQLite の全文検索機能 (FTS4/FTS5) で、DB を作ります。
トークナイザは N-Gram です。
N-Gram から元のテキストを復元するコードも書きました。
"""

from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
# import multiprocessing
import re
import unicodedata
import sqlite3
import datetime # デバッグ用
import platform # デバッグ用

def main():
    """メイン関数です。"""
    print('start\n')
    t_start = datetime.datetime.now()

    # (準備 1/6) テキストデータを用意します。
    # 検索で困りそうな文字も入れてみました。
    # (例) 謎の空白、全角記号、全角英数字、囲い文字、組文字(くみもじ)など。
    # こういった不揃いなテキストは、Python の機能で正規化しました。
    src_datas = [
        # [0]file [1]date [2]text
        ['file1.txt', '2021-01-15', 'あいうえお、うえお。（ Ａ １ ）①'],
        ['file2.txt', '2021-01-16', 'かきくけこ、くけこ。（ Ｂ ２ ）㈱'],
        ['file3.txt', '2021-01-17', 'さしすせそ、すせそ。（ Ｃ ３ ）㍿'],
        ['file4.txt', '2021-01-18', 'たちつてと、つてと。（ Ｄ ４ ）㈲'],
        ['file5.txt', '2021-01-19', 'なにぬねの、ぬねの。（ Ｅ ５ ）㌫'],
        ['file6.txt', '2021-01-20', ' た ちつて  と 、 つて と。（ Ｘ ６ ）㈹ '],
    ]
    print('(src_datas)')
    for src_data in src_datas:
        print(src_data)

    # (準備 2/6) SQLite DB のファイルを決めます。
    fts_db = ':memory:' # デバッグ用
    # fts_db = Path(r'F:\project\fts.db')
    print(f'\n(fts_db) {fts_db}')

    # (準備 3/6) データベースに接続します。
    # conn は connection の略です。
    conn = sqlite3.connect(fts_db)

    # (準備 4/6) カーソルを取得します。
    # c は cursor の略です。
    c = conn.cursor()

    # (デバッグ) Python のバージョンを表示します。
    print(f'(Python version) {platform.python_version()}')

    # (デバッグ) SQLite のバージョンを表示します。
    # SQL 文で取得することもできましたし、
    # Python の sqlite3 の定数から取得することもできました。
    c.execute('SELECT sqlite_version()')
    (sqlite_version,) = c.fetchone()
    print(f'(SQLite version) {sqlite_version}')
    print(f'(sqlite3.sqlite_version) {sqlite3.sqlite_version}')

    # # (お好みで) ジャーナルモードを MEMORY に変更します。
    # # 時間のかかる disk I/O を減らすことができました。
    # c.execute('PRAGMA journal_mode = MEMORY')

    # (準備 5/6) 『rowid を管理するテーブル』を作ります。
    # (目的) fts テーブルで、行の重複を避けるために使いました。
    # あと、行の削除を高速化するためにも使いました。
    # (理由) fts4/fts5 の仮想テーブルでは、PRIMARY KEY と
    # UNIQUE と CREATE INDEX が、使用できませんでした。
    # なので、単純に INSERT していたら重複が発生しましたし、
    # 不要になった行の DELETE にも、時間がかかりました。
    # それを解決するために、管理するテーブルを作りました。
    # (rowid を固定) 『rowid INTEGER PRIMARY KEY』を指定します。
    # これで、VACUUM しても rowid を維持することができました。
    c.execute('''CREATE TABLE IF NOT EXISTS infos(
        rowid INTEGER PRIMARY KEY,
        file TEXT UNIQUE,
        date TEXT,
        len_text INTEGER
        )''')

    # (準備 6/6) 全文検索 (full-text search, fts) をするための
    # 『仮想テーブル (VIRTUAL TABLE)』を作ります。
    # デバッグ用として、file と date の列も入れておきました。
    # file と date の MATCH 検索は必要ないので、
    # notindexed とか UNINDEXED を付けておきました。
    # これで、MATCH はしなくなりましたが、= とか LIKE での検索は、
    # 通常通りできました。もちろん、高速ではなかったです。

    if True: # True: FTS5 を使用。False: FTS4 を使用。
        # [FTS5 を使う場合]
        print('(FTS version) FTS5\n')
        c.execute('''CREATE VIRTUAL TABLE
            IF NOT EXISTS fts USING fts5(
            file UNINDEXED,
            date UNINDEXED,
            text,
            tokenize="unicode61 remove_diacritics 2"
            )''')
        # (補足) rowid 列を書く必要は無かったです。
        # VACUUM などで rowid が変わることもなかったです。
    else:
        # [FTS4 を使う場合]
        print('(FTS version) FTS4\n')
        c.execute('''CREATE VIRTUAL TABLE
            IF NOT EXISTS fts USING fts4(
            file,
            date,
            text,
            tokenize=unicode61 "remove_diacritics=2",
            notindexed=file,
            notindexed=date
            )''')
        # (補足) rowid 列を書く必要は無かったです。
        # VACUUM などで rowid が変わることもなかったです。

    # (トークナイズの説明)
    # fts4 の tokenize=unicode61 "remove_diacritics=2" と、
    # fts5 の tokenize="unicode61 remove_diacritics 2" には、
    # アクセント記号の有無 (A a À à Â â) といった、微妙な
    # 違いを無視して検索できるようにする効果がありました。

    # (実行)
    add_texts(c, src_datas)

    # (デバッグ)
    # 作った DB で全文検索をしてみます。
    test_full_text_search(cursor=c)

    # (終了 1/2) カーソル (cursor) を閉じます。
    c.close()

    # (終了 2/2) 接続 (connection) を閉じます。
    conn.close()

    # (デバッグ)
    delta = datetime.datetime.now() - t_start
    delta_str = str(delta)
    print(f'\n経過時間 {delta_str} [時:分:秒.マイクロ秒]')

    # # マイクロ秒 .000000 を除去します。
    # delta_str = delta_str.rsplit('.', maxsplit=1)[0]
    # print(f'\n経過時間 {delta_str} [時:分:秒]')

    print('end')
    return

add_texts()

テキストを仮想テーブルに追加するところです。

自分で用意した『rowid 管理用のテーブル』を使って、重複をスキップしつつ、仮想テーブルに INSERT していきます。

def add_texts(c, src_datas):
    """全文検索用の仮想テーブルに、テキストを追加していきます。"""
    # 引数リストを作ります。
    # マルチプロセス処理のところで使用します。
    args = []
    for (file, date, text) in src_datas:
        # (重複防止) もし file が DB に追加済みなら、スキップします。
        c.execute('SELECT rowid FROM infos WHERE file=?', (file,))
        row = c.fetchone()
        if row is not None:
            continue

        # 引数リストに追加します。
        args.append((file, date, text))

    # 引数リストの数を取得します。
    len_args = len(args)

    # 引数リストが空なら終了します。
    if len_args == 0:
        print(f'新規の file がありませんでした。')
        print(f'len_args: {len_args}')
        return
    else:
        print(f'(テキスト追加) {len_args} 件\n')

    # 並列数を決めます。
    MAX_WORKERS = 2
    print(f'(実行) max_workers={MAX_WORKERS}')
    # 実際には、メインプロセスは、結果をどんどん受け取って、
    # どんどん SQLite DB に追加していきました。なので、
    # メインプロセス(1) + 子プロセス(MAX_WORKERS) の数で、
    # 処理が進みました。

    # 『並列処理』のために、プロセスプールを作ります。
    # [concurrent.futures を使う場合]
    with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
        # 『関数』と『引数リスト』を渡して、実行します。
        results = executor.map(mp_func_star, args) # 戻り値は args の順番通り
        # 戻り値は『ジェネレーター』になっていました。
        # (例) type(results): <class 'generator'>

    # # [multiprocessing を使う場合]
    # with multiprocessing.Pool(processes=MAX_WORKERS) as pool:
    #     # results = pool.imap(mp_func_star, args) # 戻り値は args の順番通り
    #     results = pool.imap_unordered(mp_func_star, args) # 順番バラバラ
    #     # 戻り値は『イマップイテレーター』などになっていました。
    #     # (例) type(results): <class 'multiprocessing.pool.IMapIterator'>
    #     # type(results): <class 'multiprocessing.pool.IMapUnorderedIterator'>

        # results から『戻り値』を取り出します。
        # 戻り値は『非同期(ひどうき)』で受け取ることができました。
        # 子プロセスが完了するたびに、for 文が進んでいきました。
        n = 0
        for (n, (file, date, bigram_text)) in enumerate(results, start=1):
            # (重複防止) もし file が追加済みなら、スキップします。
            c.execute('SELECT rowid FROM infos WHERE file=?', (file,))
            row = c.fetchone()
            if row is not None:
                print(f'skipped 追加済み n: {n} {file}')
                continue

            # (デバッグ) 進捗表示
            print(f'{n}/{len_args} {file}')

            # infos テーブルに INSERT します。このとき、
            # SQLite が rowid を自動的に付けてくれました。
            c.execute(
                'INSERT INTO infos(file,date,len_text) VALUES (?,?,?)',
                (file, date, len(bigram_text)))

            # 今 INSERT したデータの rowid を取得します。
            # last_insert_rowid() は、SQLite の組み込み関数です。
            # 最後に INSERT した行の rowid を取得することができました。
            c.execute('SELECT last_insert_rowid()')
            (rowid,) = c.fetchone()

            # rowid を付けて、fts 仮想テーブルに INSERT します。
            c.execute(
                'INSERT INTO fts(rowid,file,date,text) VALUES (?,?,?,?)',
                (rowid, file, date, bigram_text))

            # 時々コミットします。
            if n % 10000 == 0:
                print(f'\n{n}/{len_args} コミット中...')
                c.connection.commit()
                # (補足) カーソルを取得した時の接続 conn は、
                # カーソルの connection 属性からも参照する
                # ことができました。

        # for 文が完了しました。
        # 残りの分をコミットします。
        print(f'\n{n}/{len_args} (last) コミット中...')
        c.connection.commit()
    return

mp_func_star()

マルチプロセス処理で、複数の引数ひきすうを渡すときの、お決まりの関数です。

def mp_func_star(args):
    """スター演算子 * で、引数をアンパックします。"""
    return mp_func(*args)

mp_func()

テキストの抽出 ⇒ 正規化 ⇒ N-Gram 変換をする関数です。

マルチプロセス処理で、どんどん実行します。

def mp_func(file, date, text):
    """
    マルチプロセスで実行する関数です。
    (1/3) ファイルからテキストを抽出して、
    (2/3) テキストを正規化して、
    (3/3) N-Gram に変換します。
    """
    # (1/3) ファイルからテキストを抽出します。
    # [テキスト抽出に使用できたライブラリ]
    # HTML [lxml.html, BeautifulSoup(markup, 'lxml') など]
    # XHTML [lxml.etree, BeautifulSoup(markup, 'xml') など]
    # XBRL [lxml.etree, BeautifulSoup(markup, 'xml'), Arelle など]
    # PDF [Xpdf command line tools の pdftotext.exe など]
    pass # 今回は text で受け取ったテキストを使用します。

    # (2/3) テキストを正規化します。
    # 検索しやすいように、文字の種類を統一します。
    # 空白も消しました。やり方はいろいろあると思います。

    # [正規化 1] 文字の種類を揃えます。
    text = unicodedata.normalize('NFKC', text)
    # これで、以下のような変換ができました。
    # 全角スペース(\u3000) -> 半角スペース(\u0020, \x20)
    # \xa0 (\u00a0, non-breaking space, nbsp) -> 半角スペース
    # 全角英数字 -> 半角英数字
    # 半角ｶﾅ -> 全角カナ
    # ① -> 1
    # ㈱ -> (株)
    # 『か』+『゛』の結合文字 -> 1文字の『が』

    # [正規化 2] 空白文字とその連続を、1 個の半角スペースにします。
    # (空白文字の説明)
    # \x20 : 半角スペース
    # \t : 水平タブ \x09
    # \n : 改行 \x0a
    # \r : 復帰 \x0d
    # \x0b : 垂直タブ \v
    # \x0c : 改頁 \f
    text = re.sub(r'(?:\x20|\t|\n|\r|\x0b|\x0c)+', '\x20', text)

    # [正規化 3] 『テキストの両端』と『全角文字のまわり』の
    # 半角スペースを消します。
    # ところで、『全角文字のまわり』についてです。
    # 日本語の文字の範囲を『正確』に調べるのは、困難でした。
    # そこで、代わりに『半角英数字でない』文字のまわりの
    # スペースを消しました。
    # (例) ' 監 査 報 告 書 ' ⇒ '監査報告書'
    text = re.sub(
        r'(?:^\x20|\x20$|(?<=[^a-zA-Z0-9])\x20|\x20(?=[^a-zA-Z0-9]))',
        '',
        text,
        )
    # (?<=...) : 後読みアサーション (あとよみアサーション)
    # (?=...) : 先読みアサーション (さきよみアサーション)

    # (3/3) テキストを N-Gram にします。
    # 自分は『バイグラム (bigram, 2-Gram)』を使用しました。
    # (例) '決算分析' -> '決算 算分 分析'
    bigram_text = bigram(text)
    # N-Gram は、普通のテキストに戻すこともできました。
    # (例) '決算 算分 分析' -> '決算分析'

    # N-Gram を返します。
    return (file, date, bigram_text)

bigram()

テキストをバイグラムにする関数です。

N-Gram の 2 文字バージョンです。

def bigram(text):
    """
    バイグラム関数です。
    テキストをバイグラムにします。
    (例) '決算分析' -> '決算 算分 分析'
    """
    if len(text) >= 2:
        # 先頭から、
        # 2 文字取得⇒ 1 文字進む⇒ 2 文字取得⇒...
        # を繰り返します。最後に、半角スペースで連結します。
        return ' '.join((text[i:i+2] for i in range(0, len(text)-1, 1)))
    return text

bigram2text()

バイグラムから、元のテキストを復元する関数です。

自分が試した限りでは、N-Gram は、２文字の場合でも (bigram)、３文字の場合でも (trigram)、元のテキストに復元することができました。

なので、N-Gram に変換する前のテキストを、保存しておく必要がなくなりました。

def bigram2text(text):
    """
    バイグラム トゥー テキスト関数です。
    バイグラムをテキストに戻します(復元します)。
    (例) '決算 算分 分析' -> '決算分析'
    """
    if len(text) > 2:
        # 先頭から、
        # 1 文字取得⇒ 3 文字進む⇒ 1 文字取得⇒...
        # を繰り返します。あと、末尾の 1 文字 text[-1] が
        # 残ってしまったので、それも取得します。
        # 最後に、1 つの文字列に連結します。
        texts = [text[i] for i in range(0, len(text), 3)]
        texts.append(text[-1])
        return ''.join(texts)
    return text

test_full_text_search()

FTS 仮想テーブルのテストです。

試しに、MATCH で検索してみる関数です。

今回のコード例では、2 件ヒットするはずです。

ところで、ヒットしたキーワードをハイライトする機能は、SQLite FTS4/FTS5 の、組み込み関数の中にもありました。

ですが、当然のことながら、INSERT したときの N-Gram の状態に、ハイライトされてしまいました。

なので、自分は、SQLite のハイライト関数とスニペット関数は使わずに、Python の正規表現でハイライトしてみました。

def test_full_text_search(cursor):
    """全文検索のテストです。"""
    print('\n(デバッグ) 全文検索をしてみます。')

    # (1/8) 検索キーワードを決めます。
    keywords = ['たち', 'つてと']
    print(f'keywords: {keywords}')

    # (2/8) 検索キーワードごとにバイグラムにして、
    # さらに引用符で囲みます。
    # 最後に、半角スペース \x20 を挟んで連結します。
    # 半角スペースは、MATCH で暗黙の AND になりました。
    bigram_keywords = []
    for keyword in keywords:
        bigram_keywords.append(f'"{bigram(keyword)}"')
    bigram_keywords = '\x20'.join(bigram_keywords)
    print(f'bigram_keywords: {bigram_keywords}')

    # (3/8) MATCH を使用して検索します。
    # (例) 'SELECT * FROM fts WHERE text MATCH "たち" "つて てと"'
    # あと、大量のテキストを列挙したら困るので、
    # LIMIT 100 件くらいのヒットで、検索を打ち切ります。
    sql = 'SELECT * FROM fts WHERE text MATCH ? LIMIT 100'
    cursor.execute(sql, (bigram_keywords,))
    print(f"c.execute('{sql}', ('{bigram_keywords}',))")

    # (4/8) 検索結果を取得します。もちろん、テキストは
    # INSERT した時と同じ、N-Gram の状態になっています。
    results = cursor.fetchall()

    # 検索結果が無ければ終了します。
    if len(results) == 0:
        print('検索結果: 無し')
        return
    else:
        print(f'検索結果: {len(results)} 件')

    # (5/8) 検索結果をハイライトするための正規表現です。
    # キーワードを縦棒で区切ってから、丸かっこで囲みます。
    # (例) re_pattern: '(たち|つてと)'
    keywords = ['たち', 'つてと', 'A*']
    re_keywords = '|'.join((keyword for keyword in keywords))
    re_pattern = f'({re_keywords})'
    # 正規表現の中の丸カッコ () は、re.sub() で \1 を使うためのものです。

    # (6/8) 検索結果を列挙します。
    for (n, (file, date, bigram_text)) in enumerate(results, start=1):
        # (7/8) N-Gram から元のテキストを復元します。
        text = bigram2text(bigram_text)

        # (8/8) キーワードをハイライトします(強調表示します)。

        # # [ハイライト A] re.sub() を使用した方法 (抜粋無し & 簡単)
        # # 全体のテキストが短いときは、この方法が簡単でした。
        # # \number の \1 は、正規表現の 1 つ目の丸カッコ () の中でマッチした
        # # 文字列に変換されます。あと、
        # # 試していませんが、re.sub() で空マッチをスキップするときは、
        # # re.sub(repl=自作関数) に自作関数を渡せばできるかもしれません。
        # marked_text = re.sub(re_pattern, r'『\1』', text, flags=re.IGNORECASE)
        # print(f'[{n}] {file} {date} {marked_text}')

        # [ハイライト B] re.finditer() を使用した方法です。
        # 長いテキストから『抜粋』してハイライトします。
        temp_texts = []
        m_iter = re.finditer(re_pattern, text, flags=re.IGNORECASE)

        # m: マッチオブジェクト (re.Match object) の略です。
        for (n_match, m) in enumerate(m_iter, start=1):
            # テキストの中での『開始位置』と『終了位置』を取得します。
            m_start = m.start()
            m_end = m.end()

            # 空マッチならスキップします。
            if m_start == m_end:
                continue

            # 適当に、前後の『最大 10 文字』を含めて取得します。
            # 末尾のバックスラッシュ(円記号)は、数式を複数行に
            # 分けて書くためのものです。
            m_text = text[m_start-10:m_start] + \
                f'『{text[m_start:m_end]}』' + \
                text[m_end:m_end+10]

            # リストに追加します。
            temp_texts.append(m_text)

            # 適当に、『最大 5 箇所』ハイライトしたら打ち切ります。
            if n_match >= 5:
                break

        # 検索結果のテキストを連結します。
        # 点を挟んで連結したら、それっぽい感じになりました。
        marked_text = ' ... '.join(temp_texts)
        print(f'[{n}] {file} {date} {marked_text}')
    return

if name == 'main': main()

main() を呼び出すための、お決まりのコードです。

Python ファイル .py の最後のところに書きます。

if __name__ == '__main__':
    main()

実行結果

FTS5 で、全文検索用の SQLite DB を作ったときの結果です。

MATCH を使った検索結果で、期待通り 2 件ヒットしました。

成功です。

start

(src_datas)
['file1.txt', '2021-01-15', 'あいうえお、うえお。（ Ａ １ ）①']
['file2.txt', '2021-01-16', 'かきくけこ、くけこ。（ Ｂ ２ ）㈱']        
['file3.txt', '2021-01-17', 'さしすせそ、すせそ。（ Ｃ ３ ）㍿']        
['file4.txt', '2021-01-18', 'たちつてと、つてと。（ Ｄ ４ ）㈲']        
['file5.txt', '2021-01-19', 'なにぬねの、ぬねの。（ Ｅ ５ ）㌫']        
['file6.txt', '2021-01-20', ' た ちつて  と 、 つて と。（ Ｘ ６ ）㈹ ']

(fts_db) :memory:
(Python version) 3.8.6
(SQLite version) 3.32.3
(sqlite3.sqlite_version) 3.32.3
(FTS version) FTS5

(テキスト追加) 6 件

(実行) max_workers=2
1/6 file1.txt
2/6 file2.txt
3/6 file3.txt
4/6 file4.txt
5/6 file5.txt
6/6 file6.txt

6/6 (last) コミット中...

(デバッグ) 全文検索をしてみます。
keywords: ['たち', 'つてと']
bigram_keywords: "たち" "つて てと"
c.execute('SELECT * FROM fts WHERE text MATCH ? LIMIT 100', ('"たち" "つて てと"',))
検索結果: 2 件
[1] file4.txt 2021-01-18 『たち』つてと、つてと。(D ... 『つてと』、つてと。(D 4) ... 『つてと』。(D 4)(有)
[2] file6.txt 2021-01-20 『たち』つてと、つてと。(X ... 『つてと』、つてと。(X 6) ... 『つてと』。(X 6)(代)

経過時間 0:00:00.288229 [時:分:秒.マイクロ秒]
end

SQLite DB を最適化する方法

SQLite で作った FTS 仮想テーブルは、'optimize' コマンドで最適化することができました。

VACUUM も、もちろん使うことができました。

どちらも時間のかかる処理でしたが、確かに、『検索速度の改善』とデータベースの『ファイルサイズ削減』に、効果がありました。

FTS 仮想テーブルを最適化するコード例

FTS4/FTS5 の仮想テーブルを optimize する Python コード例です。

import sqlite3

conn = sqlite3.connect(r'F:\project\fts.db')
c = conn.cursor()

# (最適化) VACUUM と同じくらいの時間がかかる可能性がある
# けれど、『クエリの高速化』と『ディスクの使用量削減』が
# できるかもしれないとのことでした。
print('最適化 optimize 中...')
c.execute('INSERT INTO fts(fts) VALUES("optimize")')
print('optimize 完了')

c.close()
conn.close()

SQLite DB を VACUUM するコード例

SQLite データベースを VACUUM する Python コード例です。

別のディスクに書き込むようにするときは、uri=True を指定しておいて、以下のようにファイルパスを書いたらできました。

import sqlite3

conn = sqlite3.connect(r'file:F:\project\fts.db', uri=True)
c = conn.cursor()

# (VACUUM) データベースが大きいときは、別のディスクに
# 書き込むように設定して、VACUUM することもできました。
print('VACUUM 中...')
# c.execute('VACUUM INTO ?', (r'file:P:\temp\fts.db',))
c.execute('VACUUM') # 同じディスクで VACUUM
print('VACUUM 完了')

c.close()
conn.close()

XBRL テキスト検索システムのスクリーンショット

SQLite の全文検索に、FastAPI (公式サイト) を組み合わせて、XBRL の文章を検索するシステムを作りました。

そのスクリーンショットです。

Web 画面を用意しただけで、一気にそれっぽくなりました。

EDINET API で取得しておいた XBRL と PDF が、とても便利に開けるようになりました。

決算分析システムとちがって、全文検索システムは、比較的簡単に作ることができました。

XBRL からテキストを抽出して、どんどん INSERT していくだけでした。

良かったです。

ところで、検索にかかった時間についてです。

大きなテキストが大量にヒットしたときは、それなりに時間がかかりました。

200 Hits 4.9 sec (fts: 0.8 sec, bigram2text: 3.9 sec, その他: 0.2 sec)

遅くなっていた原因は、N-Gram ⇒ テキストの復元処理でした。

このあたりは工夫次第ですね。

１回あたりの表示件数を減らすとか（３０件程度なら快適な速さでした）、初めから元のテキストを用意しておくとか（ディスクの空きを確保するのがちょっと大変でした）、マルチプロセス処理で復元するとか。

地道に、使いながら改善していってます。

SQLite の FTS4 に格納できるテキストの限界

（2021年12月28日追記）この記事を書いた当時は記事が長くなりすぎたので書かなったのですが、遭遇する人がいるかもしれないので、やっぱり追記しておきます。

FTS4 に格納できるテキストの量には、限界があるようでした。

SQLite の仕様なのか不具合なのかはわかりませんでした。

Python 3.8.6 の sqlite3 モジュールで SQLite FTS4 を使用した時の話です。

2-Gram にした日本語テキストを FTS4 仮想テーブルに INSERT していった場合です。

SQLite DB (.db) のファイルサイズが 600 GB ～ 1 TB に達したあたりで、INSERT 時に処理が進まなくなる現象が発生しました（数時間たっても進まなかったので止まっていると判断しました）。

ファイルサイズの限界はテキストの内容によって変わりましたが、同じテキストを INSERT していった場合は、同じタイミングで現象が発生しました。

現象発生時は、Python のコンソールには何もメッセージは出ていなくて、FTS4 仮想テーブルに INSERT したところで処理が止まっていました。

この現象が仕様なのか不具合なのかは分かりませんでした。

とりあえず、SQLite の FTS4 に格納できるテキストの量は、1 つの .db ファイル当たり、せいぜい 600 GB くらいと考えておいたほうが良さそうです（2-Gram にした日本語テキストを INSERT していった場合）。

自分の環境（当記事のコード例と同じ）では、どうしても現象が発生しました。

しかしまあ、そうは言ってもです。

自分の決算分析の用途では十分だったので、引き続き使用しています。

手持ちの EDINET XBRL のテキストを全部 FTS4 に INSERT しても、SQLite DB のファイルサイズは 90 GB くらいでした。

なぜ現象が発生したのか？

実験した時の資料が残っていないので詳細は割愛しますが、どうも FTS4 の中で使用している何らかの整数がオーバーフローしたんじゃないかと想像しています。理由ですが、問題が発生した SQLite DB を VACUUM してみたら、初めて見るエラーメッセージが出たんですよね。8桁以上はあったと思う大きな数値と、それっぽい変数名のようなものがあったので、そう思うに至りました。

けれども、SQLite は公式ドキュメントを見るに、毎回ガチガチにテストしてからリリースしている様子でしたので、「そんなことあるかな？」と今でも思ってはいます。

ちなみに「FTS5 の場合は現象を回避できるかな？」と思って当時実験したのですが、FTS5 の場合は仮想テーブルへの INSERT が遅くて時間がかかり過ぎたので、実験をあきらめました。