ウェブサイト検索

Flaskアプリケーションのパフォーマンスを最適化する方法

紹介

Flaskは、小規模から中規模のアプリケーションを構築するための軽量で柔軟なウェブフレームワークです。一般的に、シンプルな個人ブログから、REST API、SaaSプラットフォーム、eコマースウェブサイト、データ駆動型ダッシュボードなど、より複雑なアプリケーションまでのプロジェクトで使用されます。

しかし、アプリケーションがトラフィックのスケールや複雑さが増すにつれて、パフォーマンスのボトleneckに気づき始めるかもしれません。コンテンツ管理システム（CMS）、モバイルアプリのAPI、またはリアルタイムデータビジュアライゼーションツールを構築している場合、Flaskのパフォーマンスを最適化することは、応答性が高くスケーラブルなユーザーエクスペリエンスを提供するために重要になります。

このチュートリアルでは、Flaskアプリケーションのパフォーマンスを最適化するためのさまざまな技術とベストプラクティスを探ります。

前提条件

  • Ubuntuを実行しているサーバーと、sudo特権を持つ非ルートユーザー、そしてアクティブなファイアウォールがあります。この設定方法についてのガイダンスを得るためには、このリストからご自身のディストリビューションを選択し、初期サーバーセットアップガイドに従ってください。サポートされているバージョンのUbuntuを使用することを確認してください。

  • Linuxコマンドラインに対する親しみ。コマンドラインの入門やリフレッシュには、このLinuxコマンドライン入門ガイドを訪れてください。

  • Pythonプログラミングの基本的な理解。

  • UbuntuシステムにPython 3.7以上がインストールされていること。UbuntuでPythonスクリプトを実行する方法を学ぶには、「UbuntuでPythonスクリプトを実行する方法」についてのチュートリアルを参照してください。

Flask環境の設定

Ubuntu 24.04はデフォルトでPython 3を搭載しています。ターミナルを開いて、Python 3のインストールを再確認するために次のコマンドを実行してください:

root@ubuntu:~# python3 --version
Python 3.12.3

もしあなたのマシンにPython 3がすでにインストールされている場合、上記のコマンドは現在のPython 3のインストールバージョンを返します。インストールされていない場合は、以下のコマンドを実行してPython 3をインストールできます。

root@ubuntu:~# sudo apt install python3

次に、システムにpipパッケージインストーラーをインストールする必要があります。

root@ubuntu:~# sudo apt install python3-pip

一度 pip がインストールされたら、Flask をインストールしましょう。

Flaskはpipを使用してインストールします。システム内の他のパッケージとの競合を避けるために、仮想環境での実行をお勧めします。

root@ubuntu:~# python3 -m venv myprojectenv
root@ubuntu:~# source myprojectenv/bin/activate

root@ubuntu:~# pip install Flask

Flaskアプリケーションを作成する

次のステップは、FlaskアプリケーションのためにPythonコードを書くことです。新しいスクリプトを作成するには、好みのディレクトリに移動してください:

root@ubuntu:~# cd ~/path-to-your-script-directory

ディレクトリ内にいるときに、新しいPythonファイルapp.pyを作成し、Flaskをインポートします。それから、Flaskアプリケーションを初期化し、基本的なルートを作成します。

root@ubuntu:~# nano app.py

これにより、空のテキストエディタが開きます。ここに自分のロジックを書いたり、次のコードをコピーしたりしてください:

from flask import Flask, jsonify, request

app = Flask(__name__)

Simulate a slow endpoint
@app.route('/slow')
def slow():
    import time
    time.sleep(2)  # to simulate a slow response
    return jsonify(message="This request was slow!")

Simulate an intensive database operation
@app.route('/db')
def db_operation():
    # This is a dummy function to simulate a database query
    result = {"name": "User", "email": "user@example.com"}
    return jsonify(result)

Simulate a static file being served
@app.route('/')
def index():
    return "<h1>Welcome to the Sample Flask App</h1>"

if __name__ == '__main__':
    app.run(debug=True)

さあ、Flaskアプリケーションを実行しましょう:

root@ubuntu:~# flask run

次のcurlコマンドを使用してエンドポイントをテストできます:

/エンドポイントをテストする（静的コンテンツを提供します）：

root@ubuntu:~# curl http://127.0.0.1:5000/

[secondary_lebel Output]
<h1>Welcome to the Sample Flask App</h1>%

/slow エンドポイントをテストします（遅い応答をシミュレートします）：

root@ubuntu:~# time curl http://127.0.0.1:5000/db

この遅いエンドポイントを確認するために、Linuxのtimeコマンドを使用します。timeコマンドは、特定のコマンドやプログラムの実行時間を測定するために使用されます。これには3つの主な情報が提供されます:

  1. リアルタイム: コマンドの開始から終了までの実際の経過時間。
  2. ユーザー時間: ユーザーモードで費やされたCPU時間。
  3. システム時間: カーネルモードで消費されたCPU時間の量。

これにより、私たちの遅いエンドポイントによって実際にかかった時間を測定するのに役立ちます。出力は次のようになるかもしれません:

{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow  0.00s user 0.01s system 0% cpu 2.023 total

このリクエストは、遅い応答をシミュレートする time.sleep(2) コールのために、応答まで約2秒かかります。

/db エンドポイントをテストしましょう（データベース操作をシミュレートします）：

root@ubuntu:~# curl http://127.0.0.1:5000/db

{"email":"user@example.com","name":"User"}

これらのエンドポイントをcurlを使用してテストすることで、Flaskアプリケーションが正しく実行されていることと、レスポンスが期待通りであることを確認できます。

次のセクションでは、さまざまな手法を使用してアプリケーションのパフォーマンスを最適化する方法を学びます。

プロダクション向けのWSGIサーバーを使用してください。

Flaskの組み込み開発サーバーは、製品環境用に設計されていません。同時リクエストを効率的に処理するには、Gunicornのような製品向けのWSGIサーバーに切り替えるべきです。

Gunicornのインストールと設定

Gunicornをインストールしましょう

root@ubuntu:~# pip install gunicorn

4つのワーカープロセスを使用してGunicornでFlaskアプリケーションを実行します:

root@ubuntu:~# gunicorn -w 4 -b 0.0.0.0:8000 app:app

 % /Library/Python/3.9/bin/gunicorn -w 4 -b 0.0.0.0:8000 app:app
[2024-09-13 18:37:24 +0530] [99925] [INFO] Starting gunicorn 23.0.0
[2024-09-13 18:37:24 +0530] [99925] [INFO] Listening at: http://0.0.0.0:8000 (99925)
[2024-09-13 18:37:24 +0530] [99925] [INFO] Using worker: sync
[2024-09-13 18:37:24 +0530] [99926] [INFO] Booting worker with pid: 99926
[2024-09-13 18:37:25 +0530] [99927] [INFO] Booting worker with pid: 99927
[2024-09-13 18:37:25 +0530] [99928] [INFO] Booting worker with pid: 99928
[2024-09-13 18:37:25 +0530] [99929] [INFO] Booting worker with pid: 99929
[2024-09-13 18:37:37 +0530] [99925] [INFO] Handling signal: winch
^C[2024-09-13 18:38:51 +0530] [99925] [INFO] Handling signal: int
[2024-09-13 18:38:51 +0530] [99927] [INFO] Worker exiting (pid: 99927)
[2024-09-13 18:38:51 +0530] [99926] [INFO] Worker exiting (pid: 99926)
[2024-09-13 18:38:51 +0530] [99928] [INFO] Worker exiting (pid: 99928)
[2024-09-13 18:38:51 +0530] [99929] [INFO] Worker exiting (pid: 99929)
[2024-09-13 18:38:51 +0530] [99925] [INFO] Shutting down: Master

Gunicornを使用することの利点は次のとおりです:

  • 同時リクエスト処理: Gunicornは、複数のワーカープロセスを使用して、複数のリクエストを同時に処理することを可能にします。
  • 負荷分散: それは、サーバーリソースの最適な利用を確保しながら、受信リクエストをワーカープロセス全体にバランスよく分配します。
  • 非同期ワーカー: geventのような非同期ワーカーを使用すると、他のリクエストをブロックすることなく、長時間実行されるタスクを効率的に処理できます。
  • スケーラビリティ: Gunicornは、ワーカープロセスの数を増やすことで水平方向にスケールアップし、より多くの同時リクエストを処理することができます。
  • フォールトトレランス: 自動的に応答しないかクラッシュしたワーカーを置き換え、高い可用性を確保します。
  • 本番環境に準備済み: Flaskの開発サーバーとは異なり、Gunicornは本番環境に最適化されており、セキュリティ、安定性、およびパフォーマンス機能が向上しています。

Gunicornに切り替えることで、Flaskアプリケーションのスループットと応答性を大幅に向上させ、実際のトラフィックを効率的に処理できるようにすることができます。

キャッシングを有効にして負荷を軽減する

キャッシングは、冗長な処理を削減することによってFlaskのパフォーマンスを向上させる最良の方法の1つです。ここでは、Flask-Cachingを追加して/slowルートの結果をキャッシュします。

RedisでFlask-Cachingをインストールして構成する

必要なパッケージをインストールします:

root@ubuntu:~# pip install Flask-Caching redis

/slowルートにキャッシングを追加するために、app.pyを更新してください

エディターを開き、以下の内容でapp.pyファイルを更新してください:

root@ubuntu:~# nano app.py

from flask_caching import Cache

app = Flask(__name__)

Configure Flask-Caching with Redis
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_HOST'] = 'localhost'
app.config['CACHE_REDIS_PORT'] = 6379
cache = Cache(app)

@app.route('/slow')
@cache.cached(timeout=60)
def slow():
    import time
    time.sleep(2)  # Simulate a slow response
    return jsonify(message="This request was slow!")

最初のリクエストに対して/slowを送信した後、60秒以内のその後のリクエストはキャッシュから提供され、time.sleep()関数をバイパスします。これによりサーバーの負荷が軽減され、応答時間が速くなります。

注: このチュートリアルでは、Redisホストとしてlocalhostを使用しています。ただし、プロダクション環境では、DigitalOcean Managed Redisのような管理されたRedisサービスの使用をお勧めします。これにより、キャッシングニーズに対してより良いスケーラビリティ、信頼性、およびセキュリティが提供されます。このチュートリアルでは、App PlatformでのDigitalOcean Redisを使用したキャッシングについて、プロダクションレベルのアプリに統合する方法を学ぶことができます。

データがキャッシュされているか確認するために、/slow エンドポイントのために以下のコマンドを実行しましょう。

これは/slowエンドポイントへの最初のリクエストです。このリクエストが完了した後、/slowルートの結果がキャッシュされます。

root@ubuntu:~# time curl http://127.0.0.1:5000/slow

{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow  0.00s user 0.01s system 0% cpu 2.023 total

これは、60秒以内に/slowエンドポイントへのその後のリクエストです:

root@ubuntu:~# time curl http://127.0.0.1:5000/slow

{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow 0.00s user 0.00s system 0% cpu 0.015 total

データベースクエリの最適化

データベースクエリはしばしばパフォーマンスのボトルネックになることがあります。このセクションでは、SQLAlchemy と接続プーリングを使用してデータベースクエリの最適化をシミュレートします。

接続プーリングを使用したデータベースクエリのシミュレーション

まず、SQLAlchemyをインストールしましょう

root@ubuntu:~# pip install Flask-SQLAlchemy

app.pyを更新して接続プーリングを構成します

rom flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text

Simulate an intensive database operation
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_POOL_SIZE'] = 5  # Connection pool size
db = SQLAlchemy(app)

@app.route('/db1')
def db_operation_pooling():
    # Simulate a database query
    result = db.session.execute(text('SELECT 1')).fetchall()
    return jsonify(result=str(result))

さて、db1のルートに対してcurlリクエストを実行すると、次の出力が表示されるはずです:

root@ubuntu:~# curl http://127.0.0.1:5000/db1   

output
{"result":"[(1,)]"}

Flaskアプリケーションのパフォーマンスは、プロダクション環境で接続プーリングを実装することによって大幅に最適化できます。接続プーリングを使用すると、アプリケーションは各リクエストごとに新しい接続を作成するのではなく、既存のデータベース接続を再利用できます。これにより、新しい接続を確立するためのオーバーヘッドが減少し、応答時間が短縮され、スケーラビリティが向上します。

SQLALCHEMY_POOL_SIZEの設定は、プール内の接続の数を制限します。この値は、特定の要件とサーバーの能力に基づいて、運用環境で調整する必要があります。さらに、プールが満杯のときに追加の接続を許可するためのSQLALCHEMY_MAX_OVERFLOWや、接続を待つリクエストの待機時間を設定するためのSQLALCHEMY_POOL_TIMEOUTなど、他のプーリングオプションも検討することをお勧めします。

覚えておいてください、私たちの例はシンプルさのためにSQLiteを使用していますが、実際のシナリオでは、PostgreSQLやMySQLのようなより堅牢なデータベースを使用することが多いでしょう。これらのデータベースには独自の接続プール機構があり、SQLAlchemyのプーリングと組み合わせることで、さらに良いパフォーマンスを引き出すことができます。

接続プールを慎重に設定し利用することによって、Flaskアプリケーションが高負荷の下でもデータベース操作を効率的に処理できるようにし、全体的なパフォーマンスを大幅に向上させることができます。

Gzip圧縮を有効にする

レスポンスを圧縮することで、サーバーとクライアント間で転送されるデータ量を大幅に減らすことができ、パフォーマンスが向上します。

Flask-Compressのインストールと設定

Flask-compressパッケージをインストールしましょう。

root@ubuntu:~# pip install Flask-Compress

次に、圧縮を有効にするためにapp.pyを更新しましょう。

from flask_compress import Compress

This below command enables Gzip compression for the Flask app
It compresses responses before sending them to clients,
reducing data transfer and improving performance
Compress(app)

@app.route('/compress')
def Compress():
    return "<h1>Welcome to the optimized Flask app !</h1>"

これにより、500バイトを超える応答が自動的に圧縮され、大きな応答の転送時間が短縮されます。

本番環境では、Gzip圧縮により、特にHTML、CSS、JavaScriptのようなテキストベースのコンテンツにおいて、サーバーとクライアント間で転送されるデータ量を大幅に削減することができます。

データ転送のこの削減は、ページの読み込み時間を短縮し、ユーザー体験を向上させ、帯域幅コストを削減します。さらに、多くの最新のウェブブラウザは自動的にGzipのデコメッションをサポートしており、広く互換性のある最適化技術となっています。Gzip圧縮を有効にすることで、クライアント側に変更を必要とせずにFlaskアプリケーションのパフォーマンスとスケーラビリティを効果的に向上させることができます。

集中的なタスクをCeleryにオフロードする

メール送信や大規模データセットの処理のようなリソースを多く必要とする操作は、Celeryを使ってバックグラウンドタスクにオフロードするのが最適です。これにより、長時間実行されるタスクが受信リクエストをブロックするのを防ぎます。

Celeryは、時間のかかるタスクを非同期で実行できる強力な分散タスクキューシステムです。集中的な操作をCeleryにオフロードすることで、Flaskアプリケーションの応答性とスケーラビリティを大幅に向上させることができます。Celeryは、タスクをワーカープロセスに委任することで機能し、これらのプロセスは別のマシンで実行できるため、リソースの利用効率と並列処理が向上します。

セロリを使用することの主な利点は次のとおりです：

  1. ユーザーリクエストの応答時間が改善されました
  2. より良いスケーラビリティとリソース管理
  3. メインアプリケーションをブロックすることなく、複雑で時間のかかるタスクを処理する能力
  4. タスクスケジューリングと失敗したタスクの再試行のためのビルトインサポート
  5. RabbitMQやRedisなどのさまざまなメッセージブローカーとの簡単な統合

Celeryを活用することで、計算集約的またはI/Oバウンドのタスクを扱っているときでも、Flaskアプリケーションが応答性を保つことを確実にできます。

バックグラウンドタスクのためにCeleryを設定する

Celeryをインストールしましょう。

root@ubuntu:~# pip install Celery

次に、非同期タスクのためにCeleryを設定するためにapp.pyを更新しましょう:

from celery import Celery

celery = Celery(app.name, broker='redis://localhost:6379/0')

@celery.task
def long_task():
    import time
    time.sleep(10)  # Simulate a long task
    return "Task Complete"

@app.route('/start-task')
def start_task():
    long_task.delay()
    return 'Task started'

別のターミナルで、Celeryワーカーを起動します:

root@ubuntu:~#  celery -A app.celery worker --loglevel=info

------------- celery@your-computer-name v5.2.7 (dawn-chorus)
--- ***** ----- 
-- ******* ---- Linux-x.x.x-x-generic-x86_64-with-glibc2.xx 2023-xx-xx
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         app:0x7f8b8c0b3cd0
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

[tasks]
  . app.long_task

[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] Connected to redis://localhost:6379/0
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] mingle: searching for neighbors
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] mingle: all alone
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] celery@your-computer-name ready.

今、curl コマンドを実行して /start-task ルートにアクセスします。出力は次のようになります:

root@ubuntu:~# curl http://127.0.0.1:5000/start-task

Task started

これにより、バックグラウンドタスクがまだ実行中であるにもかかわらず、ほぼ瞬時に“タスクが開始されました”が返されます。

start_task()関数は2つのことを行います:

  • long_task.delay()を呼び出すことで、Celeryタスクが非同期的に開始されます。これは、タスクがバックグラウンドで実行されるようにキューに入ることを意味しますが、関数はその完了を待ちません。

  • すぐに文字列「タスクが開始されました」を返します。

重要な点は、実際の長時間実行されるタスク（10秒のスリープでシミュレートされています）がCeleryによって非同期に実行されるということです。Flaskのルートは、このタスクがリクエストに応答する前に完了するのを待ちません。

したがって、このエンドポイントを呼び出すと、「タスクが開始されました」という即時の応答が返されますが、実際のタスクはバックグラウンドで10秒間実行され続けます。

バックグラウンドタスクが完了した後10秒で、このログメッセージに気付くべきです:

出力はこのようになります:

[2024-xx-xx xx:xx:xx,xxx: INFO/MainProcess] Task app.long_task[task-id] received
[2024-xx-xx xx:xx:xx,xxx: INFO/ForkPoolWorker-1] Task app.long_task[task-id] succeeded in 10.xxxs: 'Task Complete'

この例は、Celeryが長時間実行されるタスクを非同期に処理することによってFlaskアプリケーションのパフォーマンスを向上させ、メインアプリケーションを応答性のある状態に保つ方法を示しています。長時間のタスクはバックグラウンドで実行されるため、Flaskアプリケーションは他のリクエストを処理することができます。

生産環境でCeleryを実装するには次のことが必要です:

  1. RabbitMQのような堅牢なメッセージブローカーを使用すること
  2. 専用の結果バックエンド（例：PostgreSQL）を使用する
  3. プロセス制御システム（例：スーパーバイザー）を使用した労働者の管理
  4. 監視ツール（例: Flower）の実装
  5. エラーハンドリングとロギングの強化
  6. タスクの優先順位付けを活用する
  7. 異なるマシン上での複数のワーカーによるスケーリング
  8. 適切な安全対策を確保すること

結論

このチュートリアルでは、さまざまなパフォーマンス向上技術を実装することによってFlaskアプリケーションを最適化する方法を学びました。これらの手順に従うことで、Flaskアプリケーションのパフォーマンス、スケーラビリティ、応答性を向上させることができ、重い負荷の下でも効率的に動作することを保証します。

関連記事: