インデックスを使用して SQL クエリを高速化する方法 [Python 編]

Python の組み込み sqlite3 モジュールを使用して SQLite データベースを操作する方法を学びます。また、クエリを高速化するためにインデックスを作成する方法も学習します。

本のページをめくっているとします。そして、探している情報をもっと早く見つけたいと考えています。どうやってそれをしたのですか？おそらく、用語索引を調べて、特定の用語を参照しているページにジャンプするでしょう。 SQL のインデックスは、書籍のインデックスと同様に機能します。

実際のシステムのほとんどでは、多数の行 (数百万行を考えてください) を含むデータベース テーブルに対してクエリを実行します。結果を取得するためにすべての行にわたるテーブル全体のスキャンが必要なクエリは非常に遅くなります。一部の列に基づいて情報を頻繁にクエリする必要があることがわかっている場合は、それらの列にデータベース インデックスを作成できます。これにより、クエリが大幅に高速化されます。

さて、今日は何を学ぶのでしょうか？ sqlite3 モジュールを使用して、Python で SQLite データベースに接続し、クエリを実行する方法を学びます。また、インデックスを追加する方法と、インデックスを追加することでパフォーマンスがどのように向上するかを確認します。

このチュートリアルに沿ってコードを作成するには、作業環境に Python 3.7 以降と SQLite がインストールされている必要があります。

注: このチュートリアルの例とサンプル出力は、Ubuntu LTS 22.04 上の Python 3.10 および SQLite3 (バージョン 3.37.2) を対象としています。

Python でデータベースに接続する

組み込みの sqlite3 モジュールを使用します。クエリの実行を開始する前に、次のことを行う必要があります。

• データベースに接続する 

• クエリを実行するデータベース カーソルを作成する

データベースに接続するには、sqlite3 モジュールの connect() 関数を使用します。接続を確立したら、接続オブジェクトに対して cursor() を呼び出して、次のようにデータベース カーソルを作成できます。

import sqlite3

# connect to the db
db_conn = sqlite3.connect('people_db.db')
db_cursor = db_conn.cursor()

ここでは、データベース people_db への接続を試みます。データベースが存在しない場合は、上記のスニペットを実行すると sqlite データベースが作成されます。

テーブルの作成とレコードの挿入

次に、データベースにテーブルを作成し、レコードを入力します。

people_db データベースに、次のフィールドを持つ people という名前のテーブルを作成しましょう。

• 名前

• 電子メール
• 仕事

# main.py
...
# create table
db_cursor.execute('''CREATE TABLE people (
                  id INTEGER PRIMARY KEY,
                  name TEXT,
                  email TEXT,
                  job TEXT)''')


...

# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

Faker による合成データの生成

次に、テーブルにレコードを挿入する必要があります。これを行うには、pip を通じてインストールできる合成データ生成用の Python パッケージである Faker を使用します。

$ pip install faker

Faker をインストールした後、Faker クラスを Python スクリプトにインポートできます。

# main.py
...
from faker import Faker
...

次のステップでは、レコードを生成して people テーブルに挿入します。インデックスによってクエリがどのように高速化されるかを理解するために、大量のレコードを挿入してみましょう。ここでは、100K レコードを挿入します。 num_records 変数を 100000 に設定します。

次に、次のことを行います。

• Faker オブジェクト fake をインスタンス化し、再現性を得るためにシードを設定します。 

• fake オブジェクトで first_name() と last_name() を呼び出して、姓名を使用して名前文字列を取得します。
• domain_name() を呼び出して偽のドメインを生成します。
• 姓名とドメインを使用して電子メール フィールドを生成します。
• job() を使用して、個々のレコードごとにジョブを取得します。

レコードを生成して people テーブルに挿入します。

# create and insert records
fake = Faker() # be sure to import: from faker import Faker 
Faker.seed(42)

num_records = 100000

for _ in range(num_records):
    first = fake.first_name()
    last = fake.last_name()
    name = f"{first} {last}"
    domain = fake.domain_name()
    email = f"{first}.{last}@{domain}"
    job = fake.job()
    db_cursor.execute('INSERT INTO people (name, email, job) VALUES (?,?,?)', (name,email,job))

# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

main.py ファイルには次のコードが含まれています。

# main.py
# imports
import sqlite3
from faker import Faker

# connect to the db
db_conn = sqlite3.connect('people_db.db')
db_cursor = db_conn.cursor()

# create table
db_cursor.execute('''CREATE TABLE people (
                  id INTEGER PRIMARY KEY,
                  name TEXT,
                  email TEXT,
                  job TEXT)''')


# create and insert records
fake = Faker()
Faker.seed(42)

num_records = 100000

for _ in range(num_records):
    first = fake.first_name()
    last = fake.last_name()
    name = f"{first} {last}"
    domain = fake.domain_name()
    email = f"{first}.{last}@{domain}"
    job = fake.job()
    db_cursor.execute('INSERT INTO people (name, email, job) VALUES (?,?,?)', (name,email,job))

# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

このスクリプトを 1 回実行して、テーブルに num_records 個のレコードを入力します。

データベースのクエリ

100K レコードを含むテーブルができたので、people テーブルに対してサンプル クエリを実行してみましょう。 

次の目的でクエリを実行してみましょう。

• 役職が「プロダクトマネージャー」であるレコードの名前と電子メールを取得します。

• クエリ結果を 10 レコードに制限します。

time モジュールのデフォルト タイマーを使用して、クエリのおおよその実行時間を取得します。

# sample_query.py

import sqlite3
import time

db_conn = sqlite3.connect("people_db.db")
db_cursor = db_conn.cursor()

t1 = time.perf_counter_ns()

db_cursor.execute("SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;")

res = db_cursor.fetchall()
t2 = time.perf_counter_ns()

print(res)
print(f"Query time without index: {(t2-t1)/1000} us")

出力は次のとおりです。

Output >>
[
    ("Tina Woods", "Tina.Woods@smith.com"),
    ("Toni Jackson", "Toni.Jackson@underwood.com"),
    ("Lisa Miller", "Lisa.Miller@solis-west.info"),
    ("Katherine Guerrero", "Katherine.Guerrero@schmidt-price.org"),
    ("Michelle Lane", "Michelle.Lane@carr-hardy.com"),
    ("Jane Johnson", "Jane.Johnson@graham.com"),
    ("Matthew Odom", "Matthew.Odom@willis.biz"),
    ("Isaac Daniel", "Isaac.Daniel@peck.com"),
    ("Jay Byrd", "Jay.Byrd@bailey.info"),
    ("Thomas Kirby", "Thomas.Kirby@west.com"),
]

Query time without index: 448.275 us

コマンド ラインで sqlite3 db_name を実行して、SQLite コマンド ライン クライアントを呼び出すこともできます。

$ sqlite3 people_db.db
SQLite version 3.37.2 2022-01-06 13:25:41
Enter ".help" for usage hints.

インデックスのリストを取得するには、.index を実行します。

sqlite> .index

現在インデックスがないため、インデックスはリストされません。

次のようにクエリ プランを確認することもできます。

sqlite> EXPLAIN QUERY PLAN SELECT name, email FROM people WHERE job='Product Manager' LIMIT 10;
QUERY PLAN
`--SCAN people

ここでのクエリ プランはすべての行をスキャンすることになっており、非効率的です。

特定の列にインデックスを作成する

特定の列にデータベース インデックスを作成するには、次の構文を使用できます。

CREATE INDEX index-name on table (column(s))

特定の役職を持つ個人の記録を頻繁に検索する必要があるとします。 job 列にインデックス people_job_index を作成すると便利です。

# create_index.py

import time
import sqlite3

db_conn = sqlite3.connect('people_db.db')

db_cursor =db_conn.cursor()

t1 = time.perf_counter_ns()

db_cursor.execute("CREATE INDEX people_job_index ON people (job)")

t2 = time.perf_counter_ns()

db_conn.commit()

print(f"Time to create index: {(t2 - t1)/1000} us")


Output >>
Time to create index: 338298.6 us

インデックスの作成にはこれほど時間がかかりますが、操作は 1 回限りです。複数のクエリを実行する場合でも、大幅な速度向上が得られます。

SQLite コマンドライン クライアントで .index を実行すると、次の結果が得られます。

sqlite> .index
people_job_index

インデックスを使用したデータベースのクエリ

クエリ プランを見ると、job 列のインデックス people_job_index を使用して people テーブルを検索していることがわかります。

sqlite> EXPLAIN QUERY PLAN SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;
QUERY PLAN
`--SEARCH people USING INDEX people_job_index (job=?)

Sample_query.py を再実行できます。 print() ステートメントのみを変更し、クエリの実行にかかる時間を確認します。

# sample_query.py

import sqlite3
import time

db_conn = sqlite3.connect("people_db.db")
db_cursor = db_conn.cursor()

t1 = time.perf_counter_ns()

db_cursor.execute("SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;")

res = db_cursor.fetchall()
t2 = time.perf_counter_ns()

print(res)
print(f"Query time with index: {(t2-t1)/1000} us")

出力は次のとおりです。

Output >>
[
    ("Tina Woods", "Tina.Woods@smith.com"),
    ("Toni Jackson", "Toni.Jackson@underwood.com"),
    ("Lisa Miller", "Lisa.Miller@solis-west.info"),
    ("Katherine Guerrero", "Katherine.Guerrero@schmidt-price.org"),
    ("Michelle Lane", "Michelle.Lane@carr-hardy.com"),
    ("Jane Johnson", "Jane.Johnson@graham.com"),
    ("Matthew Odom", "Matthew.Odom@willis.biz"),
    ("Isaac Daniel", "Isaac.Daniel@peck.com"),
    ("Jay Byrd", "Jay.Byrd@bailey.info"),
    ("Thomas Kirby", "Thomas.Kirby@west.com"),
]

Query time with index: 167.179 us

クエリの実行に約 167.179 マイクロ秒かかることがわかります。

パフォーマンスの向上

このサンプル クエリでは、インデックスを使用したクエリの方が約 2.68 倍高速です。また、実行時間は 62.71% 高速化されました。 

さらにいくつかのクエリ (ジョブ列のフィルタリングを含むクエリ) を実行して、パフォーマンスの向上を確認することもできます。 

また、注: ジョブ列にのみインデックスを作成したため、他の列を含むクエリを実行している場合、クエリはインデックスを使用しない場合よりも高速に実行されません。

まとめと次のステップ

このガイドが、頻繁にクエリされる列にデータベース インデックスを作成するとクエリが大幅に高速化される仕組みを理解していただければ幸いです。データベースのインデックスについての紹介です。複数列のインデックス、同じ列に対する複数のインデックスなどを作成することもできます。 

このチュートリアルで使用されているすべてのコードは、この GitHub リポジトリにあります。コーディングを楽しんでください!