プログラム開発の中で、性能向上・パフォーマンス向上が必要になる場面が多々あります。自分が実務の中で経験的に知った観点を整理します。（あくまでも自分の実務の中で得た経験的知識であり、必ず性能が上がるというわけではない）
※本記事での性能・パフォーマンスは、コードの処理にかかる時間のことを指しています。
高性能・ハイパフォーマンス=処理が速く終わること
ログからボトルネックを見つける
大前提として、コードのどの部分がボトルネックになっているのかを確認する必要がある。ボトルネックの種類としては、大まかに２種類のボトルネックがあるという観点をもっていると考えやすい。
- CPUバウンド：ファイルの構造が複雑であったり、値の変換処理が複雑だったり etc（経験的にこれが原因の場合、コード修正が必要になるし時間かかる）
- I/Oバウンド：ファイルのDL/UP, DBのクエリ処理（経験的にクエリ処理が問題になる場合が多い）
時間を確認する確認手段の一つとして、ログに処理にかかった時間を表示して時間のかかる処理を見つける方法がある。ログを設置するポイントとしては、上記２つの処理に関連するコードに狙いを定めて設置するのがよい。
# pythonで ETL処理を例にする
## ETL対象ファイルのDL
logging.info("ファイルDLが開始")
dl_start = time.perftime()
file = ...(ファイルのダウンロード処理・読取）...
dl_end = time.perftime()
logging.info(f"ファイルDLが終了:かかった時間は {dl_end - dl_start}")
## ファイルの中を解析する処理
logging.info("ファイル解析が開始")
analysis_start = time.perftime()
result = ...(ファイルの中を解析する処理）...
analysis_end = time.perftime()
logging.info(f"ファイル解析が終了:かかった時間は {analysis_end - analysis_start}")
## DBテーブルへ登録
logging.info("DB登録処理が開始")
query_start = time.perftime()
query_result = ...(DBにアクセスしてinsertやupdateやdeleteなど)...
query_end = time.perftime()
logging.info(f"DB登録処理が開始:かかった時間は {query_end - query_start}")
この確認ステップを飛ばすと、かけた労力が無駄になる場合があるので注意。
- CPUバウンド（ファイル解析が足を引っ張っている）のに、I/Oバウンド向けのパフォーマンス向上修正をしてしまい、大した性能向上が見られなかった。
CPUバウンドの場合
コードのリファクタリングが有効。リファクタリングでだめならサーバのスペックアップしたほうが早い。
観点①：無駄な検索を繰り返していないか？
ファイルの中身の検索などを、2回・3回…と繰り返す処理がふくまれていると、O(N^2), O(N^3)とかになる。ファイルのデータ量が多くなった際に遅くなりやすい。
# users, ordersがファイルから取得した内容と仮定。
# 各オーダーをしたユーザ名を表示したい場合を考える。
## users.json
users = [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 3, "name": "Charlie"},
]
## orders.json
orders = [
{"order_id": 101, "user_id": 2},
{"order_id": 102, "user_id": 1},
{"order_id": 103, "user_id": 3},
]
## ファイル量とともに遅くなる場合。各orderに対してusers全体から検索しているので、O(N^2)かかる。
for order in orders: # O(N)
# 毎回 users を先頭から検索している
user = next(u for u in users if u["id"] == order["user_id"]) # O(N)
print(order["order_id"], user["name"])
検索結果を辞書型配列などに一時保管しておくと高速化につながる。ただ、メモリ使用量は増える可能性があるので注意。
## 高速化の工夫：オーダーidをキー、ユーザ名を値とする辞書型配列で管理。
user_map = {user["id"]: user for user in users}
for order in orders: # O(N)
# O(1)で取得できる
user = user_map[order["user_id"]] # O(1)
print(order["order_id"], user["name"])観点②：SQLクエリ処理を繰り返していないないか？
一般に N+1 問題と呼ばれている。
[宇宙一わかりやすい] n + 1 問題とはなにが問題なのか – Qiita
はじめに バックエンド側の開発を行っているとかならず遭遇するN+1問題。 今回は初心者向けにN+1問題はなぜ避けるべき対象とされているのか。例えを用いて分かりやすく解説していきたいと思います。 対象読者 初心者エンジニア Rails を触り…
qiita.com
for の中で毎回テーブル検索を繰り返すと、DBサーバとのやり取りを繰り返すことになり、非効率になりがち。仮にordersが1万レコードあった場合を仮定すると、 I/O 関連のボトルネックが発生してしまう。
- DBとの通信が1万回発生する
- SQLの解析実行を1万回繰り返す
- ネットワーク往復が1万回発生する
# 例） order の各レコードから、注文者の名前を表示したい場合
orders = (sqlでordersテーブルを検索。 select order_id, user_id from orders)
# 各レコードに対し、注文者名を表示
for order in orders:
# 各orderごとにユーザ名検索を繰り返す
# select user_name from users where user_id = (order["user_id"])
user = (sqlでusersテーブルから user_id に紐づくレコードを検索。
select user_name FROM users WHERE users.user_id = order["user_id"]) )
print(order["order_id"], user["user_name"])sqlテーブルの中で 内部結合などで1回のクエリで済むようにするとよい。（レコード数が大きいと内部結合に負荷がかかるため、注意が必要ではある）
results = (sqlでordersテーブルとusersテーブルを内部構造して検索。
select order_id, user_name from orders
inner join users on orders.user_id = users.user_id)
# 内部結合でのクエリ結果に対し、注文者名を表示
for result in results:
print(order["order_id"], user["user_name"])観点③：並列処理にできるか？
処理の単位やデータを分割し、並列処理にすることでかかる時間を短くする方針案。
並列処理・並行処理を設計するときに最初に整理すべきこと
プログラムのパフォーマンス向上のため、並列処理・並行処理化する際に必要な観点や分割方針について整理ボトルネックの種類 ボトルネックが CPU バウンド or I/O バウンドのいずれかに該当した場合、並列処理にすることが有効になる。I/Oバ…
output-zakki.com
2026.07.11
I/Oバウンドの場合
観点①：並行処理化できないか？
async などで並行処理できるようにするのがよい。
import requests
urls = [
"https://example.com/file1.zip",
"https://example.com/file2.zip",
# ...
]
# urlsを先頭から逐次進める。urlが完了しないと次のurlに進めない
for url in urls:
response = requests.get(url)
with open(url.split("/")[-1], "wb") as f:
f.write(response.content)改善策を適用した場合
import asyncio
import aiohttp
urls = [
"https://example.com/file1.zip",
"https://example.com/file2.zip",
# ...
]
async def download(session, url):
async with session.get(url) as response:
content = await response.read()
with open(url.split("/")[-1], "wb") as f:
f.write(content)
async def main():
async with aiohttp.ClientSession() as session:
tasks = [
download(session, url)
for url in urls
]
# 各urlを並行して処理できる。同タイミングでurl1,url2,url3, など複数個の処理を開始できる
await asyncio.gather(*tasks)
asyncio.run(main())asyncありとなしの処理の違いの概要図
# async なしの場合
File1 ダウンロード待ち
↓
File2 ダウンロード待ち
↓
File3 ダウンロード待ち
...
# async ありの場合
File1 DL開始
File2 DL開始
File3 DL開始
...観点②：ストリーム処理にできないか？
ファイル全部読み取って処理開始ではなく、「少しずつ読み取って処理」を繰り返す方針案。
- 前）ファイル全体をDL→中身の処理を開始
- 後）ファイルの一部のみDL→中身を処理→同じファイルの別部分のみDL→…
# 前）ファイル全体をDL→中身の処理を開始 の例
# large.csv (10GB)
with open("large.csv", "r", encoding="utf-8") as f:
content = f.read() # 10GBすべて読み込む
for line in content.splitlines():
process(line)
後）ファイルの一部のみDL→中身を処理 を繰り返す方法
with open("large.csv", "r", encoding="utf-8") as f:
for line in f:
process(line)SQLクエリの高速化の方針案
対処方法の案
SQL①：インデックス付与
検索に時間がかかっている場合、検索のキーにインデックスを設定してあげることで検索性能が向上する場合がある。
インデックスを付与にすると、検索方法が先頭から順番に確認する方式から二分木で確認する方法になることで高速化になる。
CREATE TABLE users (
user_id BIGINT,
user_name STRING NOT NULL
)
# usersテーブルの user_id カラムにインデックス付与
# user_id をキーにして検索をする場合に高速化できる
CREATE INDEX idx_users_user_id
ON users (user_id);
SQL②：パーティションに分ける
検索対象の総量を減らすことにつながるので、この設定も有効。
CREATE TABLE orders (
order_id BIGINT,
order_date DATE NOT NULL,
user_id BIGINT,
amount NUMERIC
)
PARTITION BY RANGE (order_date); # パーティションの親テーブルの設定。order_dateカラムの値で分ける
CREATE TABLE orders_202607 # パーティションの子テーブル。2026年7月のorderを管理
PARTITION OF orders
FOR VALUES FROM ('2026-07-01')
TO ('2026-08-01');
CREATE TABLE orders_202608 # パーティションの子テーブル。2026年8月のorderを管理
PARTITION OF orders
FOR VALUES FROM ('2026-08-01')
TO ('2026-09-01');まとめ
特定後、ボトルネックの種類に応じてアクションが変わってくるため、なによりもボトルネックの部分を特定することが大事。
- CPU バウンド
- I/O バウンド
並列処理・並行処理については以下がわかりやすい。
リンク