FORCIA Tech BlogPublicationへの投稿
🐹

【Python】実行速度が遅くなるアンチパターンと改善レシピ7選

FORCIA Tech Blog 運営チーム
に公開
Python
高速化
アンチパターン
tech

前書き

こんにちは、フォルシアでエンジニアをしている太田と申します。
私は主にPythonを利用した開発を行っているのですが、開発内容の関係上、Pythonのコードを極力高速で実行したい瞬間があります。
そこで今回は、Pythonが遅くなってしまうようなアンチパターンと低速化の回避が可能なコードを紹介し、定量的に速度の比較結果も示したいと思います。

今回検証した環境と検証の際の条件を下記に示します。

検証した環境

Python: 3.14.0 (cpython)
OS: Linux 5.15.167.4-microsoft-standard-WSL2+

検証条件

  • コーディングのみの工夫でアンチパターンの改善を行う
  • Pythonから呼び出すライブラリは標準ライブラリのみとする
    • 実行時間計測にはtimeitを利用しています
  • 実行時間は処理5回分の平均時間とする
  • マルチスレッド化は今回検証対象外とする
    • 当たり前過ぎるかと思い対象外としました

ここで今回紹介するアンチパターンを盛りこんだコードを先に提示しておきます。
興味のある方はどこをどう改善すべきか考えてみてください。
※ブログ中のコードに関しては可読性向上のため、型ヒントを省略しています。

anti_pattern.py
import math
import time

data_source_1 = list(range(500000))
data_source_2 = list(range(500000, 1000000))

block_list_data = [x * 100 for x in range(10000)]

def slow_processing(src1, src2, block_list):
    all_data = src1 + src2
    valid_data = []
    for item in all_data:
        if item not in block_list:
            valid_data.append(item)

    queue = list(valid_data) # データのコピー作成
    processed_results = []

    while len(queue) > 0:
        val = queue.pop(0)
        root_val = math.sqrt(val)
        processed_results.append(root_val)

    total_val = 0
    for x in processed_results:
        total_val += x
    report = ""
    for i in range(1000): # 文字列結合のデモ用に先頭1000件のみ抽出
        report += str(processed_results[i]) + ","
    return total_val, report

res_slow, report_slow = slow_processing(data_source_1, data_source_2, block_list_data)

アンチパターン / 改善策

1．リストではなくセットを使う

特定の要素が集合に含まれているかを確認する（in演算子を使用する）場合、リスト（list）ではなくセット（set）を使用すべきです。
リストでの検索は、先頭から順番に要素を確認していく線形探索を行うため、データ量が増えるほど時間がかかります。
対してセットはハッシュテーブルという構造でデータを管理しており、データ量に関係なく計算位置を直接特定できるため、
リストで検索を行うより高速で検索が完了します。[1]

point1.py
import timeit


def check_exists_in_list(target):
    if target in data_list:
        return True
    return False


def check_exists_in_set(target):
    if target in data_set:
        return True
    return False


loop = 5
data_list = [i for i in range(1000000)]
data_set = set(range(1000000))
target = 500000
slow_result = timeit.timeit("check_exists_in_list(data_list)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("check_exists_in_set(data_set)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

Slow Version Time: 0.0077 sec
Fast Version Time: 0.0018 sec

2．文字列結合には+=ではなくjoinを使う

ループ処理の中で文字列を連結する際、+=演算子を使わず、リストに要素を追加してから最後に"".join(list)で結合すべきです。
Pythonの文字列は変更不可（イミュータブル）です。+=を実行するたびに、元の文字列と新しい文字列を合わせた新しいメモリ領域を確保し、中身をコピーします。
これがループ回数分繰り返されると、コピー量が雪だるま式に増え、処理速度が劇的に低下する一方、
joinは一度に必要なメモリを計算して確保するため高速です。[2]

point2.py
import timeit


def combine_string_with_operator(word_list):
    result = ""
    for word in word_list:
        result += word
    return result


def combine_string_with_join(word_list):
    temp_list = []
    for word in word_list:
        temp_list.append(word)
    result = "".join(temp_list)
    return result


loop = 5
words = ["Python", "Performance", "Tuning"] * 1000000
slow_result = timeit.timeit("combine_string_with_operator(words)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("combine_string_with_join(words)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

Slow Version Time: 0.0691 sec
Fast Version Time: 0.0467 sec

3．リストの作成にはforループより内包表記を使う

既存のデータから新しいリストを作成する場合、空リストを作ってforループでappend()するのではなく、内包表記を使用すべきです。
通常のforループでは、毎回appendという属性を検索し、メソッドとして呼び出すオーバーヘッドがかかります。
リスト内包表記はLIST_APPENDという専用のバイトコードを使用し、Python仮想マシン内でC言語レベルの最適化されたループ処理が行われるため、インタプリタの介入が減り高速化します。[3]

point3.py
import timeit


def make_list_with_for_loop(data):
    result = []
    for x in data:
        result.append(x * 2)
    return result


def make_list_with_connotation(data):
    result = [x * 2 for x in data]
    return result


loop = 5
data = range(1000000)
slow_result = timeit.timeit("make_list_with_for_loop(data)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("make_list_with_connotation(data)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

Slow Version Time: 0.0335 sec
Fast Version Time: 0.0264 sec

4．先頭要素の操作にはpop(0)ではなくpopleft()を使う

かなり有名な内容だと思います。例として先頭要素の取得を行うシチュエーションで考えてみます。
リストをキューとして扱う際、先頭要素の取得する際にlist.pop(0)を使うのは避け、collections.dequeを使用してpopleft()を行うべきです。
これは、リストはメモリ上で連続した配列として実装されており、先頭を取得すると、空いた穴を埋めるために残りの全要素を1つずつ前にずらすシフト処理が発生し低速化を招きます。
これに対しdequeは双方向リスト構造のため、先頭要素の取得もシフト処理なしで行えます。[4]
もちろん、先頭要素の削除だけ行いたい場合であればlistをreverseして末尾を取得することでもdequeと同レベル感の実行速度を実現できます。
ですが、dequeにはappendとpopleftが両立できるというメリットもあり、このあたりの選定はデータの追加と削除が必要かどうか、
インデックスアクセスが必要かどうか、扱いたいシチュエーションが先入れ先出しか後入れ先出しかなどケースバイケースになってくるかと思います。

point4.py
import timeit
from collections import deque


def get_first_content_with_pop(queue):
    item = queue.pop(0)
    return item


def get_first_content_with_popleft(queue):
    item = queue.popleft()
    return item


loop = 5
queue = [i for i in range(1000000)]
dequed_queue = deque(queue)
# このpointでは出力する桁数を変えています
slow_result = timeit.timeit("remove_first_content_with_pop(queue)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.8f} sec")
fast_result = timeit.timeit("remove_first_content_with_popleft(dequed_queue)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.8f} sec")

Slow Version Time: 0.00035998 sec
Fast Version Time: 0.00000078 sec

5．グローバル変数をローカル変数に入れる

ループ内で何度も使用するグローバル変数やモジュールの関数は、ループに入る前にローカル変数に代入（エイリアス化）して使用すべきです。
Pythonの変数解決は「ローカル」→「グローバル」→「組み込み」の順で行われます。関数内のローカル変数は配列のインデックスアクセス（LOAD_FAST）で済みますが、
グローバル変数は辞書検索（LOAD_GLOBAL）が必要です。特に回数が多いループでは、この検索コストにより低速化を招く場合があります。[5]

point5.py
import timeit
import math


def calculate_roots_with_global(n):
    res = []
    for i in range(n):
        res.append(math.sqrt(i))
    return res


def calculate_roots_with_local(n):
    res = []
    sqrt = math.sqrt
    append = res.append
    for i in range(n):
        append(sqrt(i))
    return res


loop = 5
number = 1000000
slow_result = timeit.timeit("calculate_roots_with_global(number)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("calculate_roots_with_local(number)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

Slow Version Time: 0.0458 sec
Fast Version Time: 0.0389 sec

6．組み込み関数を使う

ここにきてかなり当たり前に近い内容かもしれませんが、自分でforループなどで計算を行うのではなく、Pythonの組み込み関数を使用すべきです。
Pythonコードで書いたループは、インタプリタ上でバイトコードとして一つずつ実行されるためオーバーヘッドがあります。
一方、sumなどの組み込み関数はC言語で実装されており、ループ処理そのものがC言語レベルの速度で実行されるため高速です。[6]

point6.py
import timeit


def sum_with_for_loop(data):
    total = 0
    for x in data:
        total += x
    return total


def sum_with_built_in_functions(data):
    total = sum(data)
    return total


loop = 5
data = range(1000000)
slow_result = timeit.timeit("sum_with_for_loop(data)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("sum_with_built_in_functions(data)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

Slow Version Time: 0.0163 sec
Fast Version Time: 0.0072 sec

7．itertools.chainを使う

複数のリストを順番に処理したい場合、+演算子で新しいリストを作るのではなく、itertools.chainを使用すべきです。
例えばリストの結合にlist1 + list2を実行した際、両方のリストの要素をコピーして新しいリストを作成するため、メモリ消費とコピー時間がかかります。
itertools.chainはデータをコピーせず、リスト間をまたいで順番にアクセスするイテレータを作成するため、初期化コストがほぼゼロでメモリも消費しません。[7]

point7.py
import timeit
import itertools


def combine_list_with_operator(list1, list2):
    result_list = list1 + list2
    return result_list


def combine_list_with_chain(list1, list2):
    result_list = itertools.chain(list1, list2)
    return result_list


loop = 5
list1 = [i for i in range(500000)]
list2 = [i for i in range(500000, 1000000)]
# このpointでは出力する桁数を変えています
slow_result = timeit.timeit("combine_list_with_operator(list1, list2)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.8f} sec")
fast_result = timeit.timeit("combine_list_with_chain(list1, list2)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.8f} sec")

Slow Version Time: 0.00701807 sec
Fast Version Time: 0.00000099 sec

実演

ということで冒頭紹介したコードと上記の改善を施したコードで実行速度を比較してみます。

slow_vs_fast.py
import math
import timeit
from collections import deque
from itertools import chain

data_source_1 = list(range(500000))
data_source_2 = list(range(500000, 1000000))

block_list_data = [x * 100 for x in range(10000)]

def slow_processing(src1, src2, block_list):
    # [Point 7] 巨大なリストをコピーして新規作成している
    all_data = src1 + src2

    valid_data = []

    # [Point 3] forループでlist作成している
    for item in all_data:

        # [Point 1] itemが含まれるかlistを利用し確認している
        if item not in block_list:
            valid_data.append(item)

    processed_results = []
    while len(valid_data) > 0:

        # [Point 4] リストをキューとして使い、先頭削除ごとに全要素をシフトしている
        val = valid_data.pop(0)

        # [Point 5] ループのたびにmathモジュールからsqrtを検索している
        root_val = math.sqrt(val)
        processed_results.append(root_val)

    total_val = 0

    # [Point 6] forループで合計の計算をしている
    for x in processed_results:
        total_val += x

    # [Point 2] 文字列結合を += で行っている
    report = ""
    for i in range(1000): # 文字列結合のデモ用に先頭1000件のみ
        report += str(processed_results[i]) + ","
    return total_val, report

def fast_processing(src1, src2, block_list):
    # [Point 1] リストをセットに変換し、ハッシュ検索を可能にする
    block_set = set(block_list)

    # [Point 7] itertools.chain でイテレータとして扱う
    all_data_iter = chain(src1, src2)

    # [Point 3] リスト内包表記を使用 ※Point1のblock_setを使用
    valid_data = [item for item in all_data_iter if item not in block_set]

    # [Point 4] dequeを使用し、popleft()で削除を行う
    queue = deque(valid_data)
    processed_results = []

    # [Point 5] 何度も呼ぶメソッドをローカル変数にキャッシュ
    local_sqrt = math.sqrt
    local_append = processed_results.append

    while queue:
        val = queue.popleft()
        local_append(local_sqrt(val))

    # [Point 6] 組み込み関数sumを使用
    total_val = sum(processed_results)

    # [Point 2] リスト内包表記で文字列化し、joinで一括メモリ確保・結合、先頭1000件を取得
    # 厳密に言えば遅い関数のreportと比較して末尾にカンマがつかないという違いはあります
    report = ",".join([str(x) for x in processed_results[:1000]])
    return total_val, report


loop = 5
slow_result = timeit.timeit("slow_processing(data_source_1, data_source_2, block_list_data)", globals=globals(), number=loop)
print(f"Slow Version Time: {(slow_result / loop):.4f} sec")
fast_result = timeit.timeit("fast_processing(data_source_1, data_source_2, block_list_data)", globals=globals(), number=loop)
print(f"Fast Version Time: {(fast_result / loop):.4f} sec")

実行した結果がこちらです。

Slow Version Time: 130.5239 sec
Fast Version Time: 0.0806 sec

元のコードはアンチパターンの寄せ集めですので、この実験での定量性にそれほど大きい意味はないと考えていますが、
上記の改善を実施するだけで1000倍程度は高速になっていることがわかります。

また、どの処理にどれだけ時間がかかっているかも確認してみます。
各処理の確認にはline_profilerを利用しました。
確認結果は下記です。これを見ると最初に提示したアンチパターンのコードでは
特にPoint 4とPoint 1の内容が処理に時間がかかっている原因であることがわかります。

Timer unit: 1e-06 s
Total time: 136.578 s
File: speed_check.py
Function: slow_processing at line 12

# 行数   実行回数     処理時間 平均処理時間 処理時間割合 実行コード
Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================

    12                                           @profile
    13                                           def slow_processing(src1, src2, block_list):
    14                                               # [Point 7] 巨大なリストをコピーして新規作成している
    15         1       5758.7   5758.7      0.0      all_data = src1 + src2
    16                                           
    17         1          2.1      2.1      0.0      valid_data = []
    18                                           
    19                                               # [Point 3] forループでlist作成している
    20   1000001     271940.3      0.3      0.2      for item in all_data:
    21                                           
    22                                                   # [Point 1] itemが含まれるかlistを利用し確認している
    23   1000000   39246986.3     39.2     28.7          if item not in block_list:
    24    990000     296312.5      0.3      0.2              valid_data.append(item)
    25                                           
    26         1          1.3      1.3      0.0      processed_results = []
    27    990001     382809.9      0.4      0.3      while len(valid_data) > 0:
    28                                                   # [Point 4] リストをキューとして使い、先頭削除ごとに全要素をシフトしている
    29    990000   95019638.7     96.0     69.6          val = valid_data.pop(0)
    30                                           
    31                                                   # [Point 5] ループのたびにmathモジュールからsqrtを検索している
    32    990000     612237.4      0.6      0.4          root_val = math.sqrt(val)
    33    990000     324658.6      0.3      0.2          processed_results.append(root_val)
    34                                           
    35         1          1.0      1.0      0.0      total_val = 0
    36                                           
    37                                               # [Point 6] forループで合計の計算をしている
    38    990001     210188.6      0.2      0.2      for x in processed_results:
    39    990000     206174.3      0.2      0.2          total_val += x
    40                                           
    41                                               # [Point 2] 文字列結合を += で行っている
    42         1          1.4      1.4      0.0      report = ""
    43      1001        220.9      0.2      0.0      for i in range(1000): # 文字列結合のデモ用に先頭1000件のみ
    44      1000        621.1      0.6      0.0          report += str(processed_results[i]) + ","
    45         1          6.0      6.0      0.0      return total_val, report

Total time: 0.684865 s
File: speed_check.py
Function: fast_processing at line 48

# 行数   実行回数     処理時間 平均処理時間 処理時間割合 実行コード
Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
    48                                           @profile
    49                                           def fast_processing(src1, src2, block_list):
    50                                               # [Point 1] リストをセットに変換し、ハッシュ検索を可能にする
    51         1        211.8    211.8      0.0      block_set = set(block_list)
    52                                           
    53                                               # [Point 7] itertools.chain でイテレータとして扱う
    54         1          3.4      3.4      0.0      all_data_iter = chain(src1, src2)
    55                                           
    56                                               # [Point 3] リスト内包表記を使用 ※Point1のblock_setを使用
    57         1      25897.0  25897.0      3.8      valid_data = [item for item in all_data_iter if item not in block_set]
    58                                           
    59                                               # [Point 4] dequeを使用し、popleft()で削除を行う
    60         1       6262.9   6262.9      0.9      queue = deque(valid_data)
    61         1          2.5      2.5      0.0      processed_results = []
    62                                           
    63                                               # [Point 5] 何度も呼ぶメソッドをローカル変数にキャッシュ
    64         1          3.2      3.2      0.0      local_sqrt = math.sqrt
    65         1          2.3      2.3      0.0      local_append = processed_results.append
    66                                           
    67    990001     211779.0      0.2     30.9      while queue:
    68    990000     200828.9      0.2     29.3          val = queue.popleft()
    69    990000     234395.0      0.2     34.2          local_append(local_sqrt(val))
    70                                           
    71                                               # [Point 6] 組み込み関数sumを使用
    72         1       5136.8   5136.8      0.8      total_val = sum(processed_results)
    73                                           
    74                                               # [Point 2] リスト内包表記で文字列化し、joinで一括メモリ確保・結合、先頭1000件を取得
    75                                               # 厳密に言えば遅い関数のreportと比較して末尾にカンマがつかないという違いはあります
    76         1        338.9    338.9      0.0      report = ",".join([str(x) for x in processed_results[:1000]])
    77         1          3.4      3.4      0.0      return total_val, report

まとめ

もちろん、可読性の維持など様々な制約により改善策を自然に取り入れることが難しい場合も多々あるかとは思います。
また、今回ご紹介した方法は万能薬ではなく、場合によっては効果が見られない / 逆効果になりうるのも事実です。
例えば今回支配的だったPoint 4ですが、listをdequeする処理が入るため、listの要素数などによってはlistをdequeに変換する処理が支配的となり、逆効果になる場合もあります。
その他のPointについても他ブログで様々な議論がありますので、いくつかご紹介させていただきます。

とはいえ少しの工夫を行うことで、実行速度面でかなり改善することもあるかと思っております。本記事が何かのご参考になれば幸いです。

この記事を書いた人

太田 圭祐
2023年キャリア入社 エンジニア
こんにちは

脚注

  1. TimeComplexity - Python Wiki ↩︎

  2. PythonSpeed/PerformanceTips - String Concatenation ↩︎

  3. PythonSpeed/PerformanceTips - Loops ↩︎

  4. collections.deque - Python ドキュメント ↩︎

  5. PythonSpeed/PerformanceTips - Local Variables ↩︎

  6. Built-in Functions - Pythonドキュメント ↩︎

  7. itertools - Python ドキュメント ↩︎

FORCIA Tech Blog
FORCIA Tech BlogPublication

――「つなげれば、見える。」――膨大・複雑なデータから必要な情報を的確に探し出す検索テクノロジーを基にしたシステム開発・サービス提供並びに、コンサルティングで企業のDXを推進するフォルシア株式会社テックブログです。 エンジニア積極募集中！！

Discussion