分析ノート

前回の記事がマルチスレッドだったので今回はマルチプロセスを紹介します。

Pythonにおけるマルチプロセスの１番のメリットはGILの制約を回避できることでしょうね。

ただ、先に書いておきますが、この記事で書いている方法はJupyter notebookのセルに直接書くと正常に動作せずエラーになることがあります。.pyファイルを作成してそこに記入して使うようにしましょう。

マルチプロセスを実装するには、最近はconcurrent.futuresのProcessPoolExecutorを使います。
参考: concurrent.futures — 並列タスク実行 — Python 3.12.6 ドキュメント

ドキュメントのサンプルコードを参考に動かしてみましょう!
例として取り上げられているのは素数判定ですね。Pythonで処理が完結するのですが、GIL制約のためマルチスレッドだと高速化の恩恵が受けられないものです。

from concurrent.futures import ProcessPoolExecutor
import math


PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419
    ]

def is_prime(n):
    print(f"整数 {n} を素数判定します")
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d は素数か: %s' % (number, prime))

if __name__ == '__main__':
    main()

# 以下実行結果
"""
整数 112272535095293 を素数判定します
整数 112582705942171 を素数判定します
整数 112272535095293 を素数判定します
整数 115280095190773 を素数判定します
整数 115797848077099 を素数判定します
整数 1099726899285419 を素数判定します
112272535095293 は素数か: True
112582705942171 は素数か: True
112272535095293 は素数か: True
115280095190773 は素数か: True
115797848077099 は素数か: True
1099726899285419 は素数か: False
"""

最初にそれぞれの値の素数判定が始まってる旨のメッセージが出てその後に結果が順番に出てきたので、並行して処理されているのが確認できました。

is_prime(n)が並行して実行している処理です。

ProcessPoolExecutor() でエクゼキューターを作成して、今回は submit()ではなく、mapで適用していますね。map()には第一引数で並列実行したい関数を渡し、次の引数でその関数に渡す引数のリストを渡します。

submit と map はどちらもProcessPoolExecutor や ThreadPoolExecutor の継承元の抽象クラスのExecutor に実装されているメソッドなので、実はマルチプロセスとマルチスレッドのどちらでも両方使うことができます。お好みの方で書いたらよさそうです。

細かい挙動は異なっていて、前回のsubmit()ではas_completed()を使って終わった順番に処理を取り出していましたが、map()を使う場合は、処理自体は並列して同時に行われて順不同で完了しますが、結果の取り出しは渡した引数の順番になります。

とっくの昔に、threadingを使ったマルチスレッド処理について記事を書いていたつもりだったのに、まだ書いてないことに気づきました。(そして、マルチプロセスの処理についてもまだ書いてませんでした。)

それでは気づいたこのタイミングで記事にしようと思ったのですが、改めてドキュメントを見てみると、concurrent.futures というより高レベルなモジュールがあるとのことでしたので、こちらを利用したマルチスレッド処理について紹介します。

先に言っておきますが、PythonにはGIL (Global Interpreter Lock) という制約があって、マルチスレッドにしたとしても、Pythonインタープリタは一度に1つのスレッドしか実行できません。なので、Pythonで完結するプログラムはマルチスレッドしても高速化の恩恵はありません。では、いつマルチスレッドは使うのかというと、Python外部のリソース(ストレージとかOSの処理とかWebアクセスとか)の待ち時間が発生する場合になります。

前置きが長くなってきましたが、実際に、concurrent.futuresを使ったマルチスレッドの並列処理のサンプルコードを紹介します。concurrent.futures.ThreadPoolExecutor というのを使います。
参考: concurrent.futures.ThreadPoolExecutor

5つのサイトへのアクセスを並列でやってみましょう。

import concurrent.futures
import requests
import time


# 取得するURLのリスト
URLS = [
    'http://www.example.com',
    'http://www.python.org',
    'http://www.openai.com',
    'http://www.wikipedia.org',
    'http://www.github.com'
]


# URLからコンテンツを取得する関数
def fetch_url(url):
    print(f"実行開始: {url}")
    response = requests.get(url)
    print(f"実行完了: {url}")
    return url, response.status_code, len(response.content)


# マルチスレッドでURLを並列取得する
start_time = time.time()

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # 各URLに対してfetch_url関数を並列実行
    futures = {executor.submit(fetch_url, url): url for url in URLS}

    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        try:
            url, status, content_length = future.result()
            print(f"URL: {url}, Status: {status}, length: {content_length}")
        except Exception as e:
            print(f"{url}でエラーが発生しました: {e}")

print(f"処理時間: {time.time() - start_time}秒")

# 以下結果
"""
実行開始: http://www.example.com
実行開始: http://www.python.org
実行開始: http://www.openai.com
実行開始: http://www.wikipedia.org
実行開始: http://www.github.com
実行完了: http://www.python.org
URL: http://www.python.org, Status: 200, length: 50928
実行完了: http://www.github.com
URL: http://www.github.com, Status: 200, length: 254186
実行完了: http://www.openai.com
URL: http://www.openai.com, Status: 403, length: 14186
実行完了: http://www.example.com
URL: http://www.example.com, Status: 200, length: 1256
実行完了: http://www.wikipedia.org
URL: http://www.wikipedia.org, Status: 200, length: 78458
処理時間: 0.49734020233154297秒
"""

ドキュメントのコードをもとにしていますが、fetch_url()メソッドの最初と最後にprit文を差し込んで5つのURLについて同時に処理が進んでいるのが分かるようにしました。開始と終了が異なる順番で結果がprintされていて、並列で動いてた感がありますね。

さて、上記コードの fetch_url() がマルチスレッドで実行されていた関数本体ですが、 肝心のThreadPoolExecutorはかなり使い方にクセがあります。

oncurrent.futures.ThreadPoolExecutor(max_workers=5) でエグゼキューターを作って、submit()や、as_completed()というメソッドを使っていますね。

submit() は実行キューへタスクを送信するメソッドです。

そして、もう一つ、oncurrent.futures.as_completed() というのを使っています。
こちらは、送信された非同期タスクが完了した順にFutureオブジェクトを返すジェネレータ関数です。これを使うことで、並列で動いていたメソッドが完了した順に、後続の処理を行うことができます。
上の例では、future.result() でメソッドの戻り値を受け取って、順次printしています。

使い所は慎重に選ばないと高速化等の効果は得られないですし、書き方にクセがあるので、慣れないと少々戸惑うのですが、ハードウェアアクセスの待ち時間が長い時や外部リソースへのアクセスを伴う処理の高速化では非常に役に立つものなので機会があったら使ってみてください。

今回はStreamlitでアニメーションを作成します。

といっても、やることは以前紹介したプレースホルダーの中身を順次更新し続けるだけ、という実装です。
参考: Streamlitのコンテナを使って動的にページを表示する

アニメーションさせるためには一つの枠を連続的に書き換えて画像を表示するので、st.empty() を使います。

とりあえず一個やってみましょう。画像の描写はmatplotlibを使ってみました。お試しなのでアニメーションの内容は線分をぐるぐる回すだけです。(両端を三角関数で実装します。)

import streamlit as st
import matplotlib.pyplot as plt
import numpy as np


# 描画エリアを設定
fig = plt.figure()
ax = fig.add_subplot(111)

ax.set_xlim(-1.2, 1.2)
ax.set_ylim(-1.2, 1.2)

# アニメを描写するプレースホルダーを作成
placeholder = st.empty()

# Streamlitのアニメーション表示
for i in range(100):
    ax.clear()
    ax.set_xlim(-1.2, 1.2)
    ax.set_ylim(-1.2, 1.2)
    ax.plot(
        [np.cos(i*0.1), -np.cos(i*0.1)],
        [np.sin(i*0.1), -np.sin(i*0.1)]
    )
    
    # プレースホルダーを更新
    placeholder.pyplot(fig)

これで線がぐるぐる回るアニメーションが表示できます。

あれ、time.sleep(0.01)とかウェイトを入れておかないとこのfor文が一瞬で終わってしまうんじゃないの？と思われるかもしれませんが、実験してみたところちょうど良い感じにアニメーションになりました。

どうもstreamlitの仕様として一枚一枚の画像の表示(pyplot)にウェイトがかかっているようです。

これは結構大きなメリットで、あまり表示時間とか気にせずにいい感じのアニメーションが作れます。

一方で、time.sleep(0.01) で0.01秒間隔の表示で1000フレーム使ってピッタリ10秒の動画を作ろう！みたいな調整は困難です。まぁ、これはstreamlitは動画作成を念頭に置いたものではないと思うので仕方ないですね。

ただし、デフォルトだと動作が早すぎるという場合はtime.sleep()を使ってウェイトを増やしましょう。

2連続のPyGWalker関係の話です。そして、相変わらずStreamlit関係の記事です。

発展著しい両ライブラリですが、StreamlitにPyGWalkerを埋め込んで動かすこともできます。そして、使用感としてはJupyterよりStreamlitに埋め込んだ方が使い勝手がいいですね。

使用するメソッドですが、streamlitがPyGWalkerを埋め込むメソッドを持っているわけではなく、PyGWalker側がStreamlit上で動作するメソッドを持っているので注意が必要です。Streamlit側のドキュメントを読み込んでも本機能についての記述は出てきません。(少なくとも今日時点では。)

こちらを読みます。
参考: PyGWalkerとStreamlitを使ったデータの探索と情報共有 – Kanaries

StreamlitRenderer というのを使えば良いのですね。そして、設定を保存するspec引数もあります。

1点、Jupyterで動かす場合との違いなのですがセキュリティ上の理由なのかわかりませんがデフォルトではspecで指定したjsonファイルへの書き込み、要するに保存ができません。すでにどこかで保存されたダッシュボードの読み込みだけが可能という挙動になります。

これは、spec_io_mode 引数がデフォルトで”r” (読み込みモード) になっているためです。Streamlit上で作ったビューをそのまま保存したい場合は、”rw” を指定する必要があります。

注意点はこれだけなので、早速やってみましょう。データは何でもいいのでまたワインです。

from pygwalker.api.streamlit import StreamlitRenderer
import pandas as pd
import streamlit as st
from sklearn.datasets import load_wine

st.set_page_config(layout="wide")

# データ読み込み
wine = load_wine()
# 特徴量名を列名としてDataFrame作成
df = pd.DataFrame(
    wine.data,
    columns=wine.feature_names,
)

# target列も作成する。
df["target"] = wine.target
df["class"] = df["target"].apply(lambda x: wine["target_names"][x])

pyg_app = StreamlitRenderer(
    df,
    spec="./st_config.json",
    spec_io_mode="rw"
)
pyg_app.explorer()

一番最後の、explorer()を忘れないように注意してくださいね。

これでStreamlit上でもTableau風のUIでグラフを描けるようになりました。

Tableau public でローカルファイルセーブが実装されたのでやや存在感が薄れているのですが、TableauライクなダッシュボードをPythonで作れるPyGWalkerの記事2本目です。

前回書いたのがこのライブラリが登場した直後だったので、当時は今と比べるとまだ基本的な機能も揃っていなかったのですが、現時点では待望のダッシュボードの保存機能が実装されているのでその紹介です。

参考: PyGWalkerでデータフレームを可視化してみる

これ、使い方はすごく簡単で、walk メソッドで起動する時に、jsonファイルのパスをspec引数へ渡し、ダッシュボードを作ったら保存ボタンを押すだけです。(自動保存は今日時点ではサポートされていないらしい。

ReadMe にも記載がありますね。

前回の記事と同じようにワインのデータでやってみましょう。

import pandas as pd
from sklearn.datasets import load_wine
import pygwalker as pyg


# データ読み込み
wine = load_wine()
# 特徴量名を列名としてDataFrame作成
df = pd.DataFrame(
    wine.data,
    columns=wine.feature_names,
)

# target列も作成する。
df["target"] = wine.target
df["class"] = df["target"].apply(lambda x: wine["target_names"][x])

walker = pyg.walk(df, spec="./config.json"). # 設定の保存先をspecで指定 

こうすると、spec で指定したファイルが存在しなければ自動的に作成されます。そしてそこに設定が保存されます。起動時点で、specで指定したファイルが存在していたらそれが読み込まれて前回の続きから作業ができます。

繰り返しですが、「保存」そのものは自動ではやってくれないので、Saveのアイコンを確実に押しましょう。かなりわかりにくいですがこいつです。

Saveの文字はマウオーバーして出てきてくだけなので、その下の歯車付きテキストファイルのようなアイコンを探してください。

今回紹介するのは、st.set_page_config という設定コマンドです。

白状しておくと、僕は基本的に st.set_page_config(layout = “wide”) しか使っていません。(ほぼ確実に使うのになかなかこの文を覚えなくて何度も調べているので記事にしました。)

ドキュメントはこちらです。
参考: st.set_page_config – Streamlit Docs

Noteとして、 “This must be the first Streamlit command used on an app page, and must only be set once per page.” と書いてある通り、このメソッドは最初に1回だけ呼び出す必要があります。

これを使うと、デフォルトで中央寄席になっているレイアウトを画面全体に広げたり、ページタイトル(デフォルトではファイル名)を設定したり、ページアイコンを設定したりできます。

引数はドキュメントにある通りで以下の通りです。

page_title : ページのタイトルを文字列で指定。
page_icon: 絵文字や絵文字コードを用いてアイコンを指定できる。”random”も可能。
絵文字コードの一覧はこちら。
layout: ページのレイアウト。”centered” or “wide” の2種類。 centered がデフォルトなので、使う場合は基本的には “wide”の方を使用することになります。
initial_sidebar_state: サイドバーの初期状態。”auto”/ “expanded”/ “collapsed” から指定できる。基本的に、デバイスサイズで判断して動いてくれるデフォルトの”auto”で良いと思いますが、必ず表示したい/隠したいという場合は残り二つも選択肢になると思います。
menu_items: 右上に表示するメニューの設定です。

menu_itemsは指定する場合は次のように辞書で指定します。
下記のサンプルはドキュメントからそのまま持ってきました。

import streamlit as st

st.set_page_config(
    menu_items={
        'Get Help': 'https://www.extremelycoolapp.com/help',
        'Report a bug': "https://www.extremelycoolapp.com/bug",
        'About': "# This is a header. This is an *extremely* cool app!"
    }
)

“Get Help” / “Report a bug” はそれぞれURLを指定するとリンクになります。Noneの場合はメニューにこれらの項目自体がなくなります。

“About” はマークダウンの文字列で、ここに書いた内容がそのまま表示されるようになります。指定しない場合はStreamlitのデフォルトのテキストです。

今回はStreamlitで複数ページのアプリを作る方法を紹介します。この記事で紹介している方法を使うと、複数のファイルに分けてStreamlitのアプリを作れるのでコードの管理も簡単になります。

一応書いておくと、サイドバーに選択式のウィジェットを配置して、その選択によって表示する項目を切り分ける実装にするとこの記事に書いてる方法を使わなくても複数ページっぽく振る舞うアプリを作ることはできます。ただ、コードが長くなることや名前空間を共有してしまう不便さ等もあるので別ページのものは別ページで作った方が良いことが多いです。(逆に1ファイルにまとめた方が便利なこともありますが。)

複数ページのアプリを作る場合、最初にこちらのページからリンクされている各ページを参照してStreamlitの方針を把握しておくことをお勧めします。
参考: Multipage apps – Streamlit Docs

では具体的な方法を見ていきましょう。といっても、ほとんどのケースで一番最初のpagesディレクトリを使う方法で十分だと思います。

pagesディレクトリを使用する方法

これは Streamlitのメインページとなる.py ファイルと同じディレクトリに pages という名前でディレクトリを作り、その下に .py ファイルを作るというものです。

ドキュメントでは以下の構成が例示されていますね。

your_working_directory/
├── pages/
│   ├── a_page.py
│   └── another_page.py
└── your_homepage.py

この通りに作って動かして見ましょう。ファイルの中身は全部からっぽです。

% mkdir your_working_directory
% cd your_working_directory
% mkdir pages
% touch your_homepage.py
% touch pages/a_page.py
% touch pages/another_page.py
% streamlit run your_homepage.py

するとこのような画面になります。

実際に動かしていただくと確認できるのですが、それぞれURLが、
– http://localhost:8501/
– http://localhost:8501/a_page
– http://localhost:8501/another_page
となります。メインのファイルがルートパスになる感じですね。URL中には/pages/の文字は含まれません。

ファイル名と、上記の画面のサイドバーに表示されるページ名を見比べていただけると分かるのですが、 _ がスペースに変換されています。これは _ や – がセパレーターとして半角スペースに変換されるからです。また、拡張子 .py も取り除かれます。

また、ファイル名の先頭に正の数字をつけていた場合、それも取り除かれます。項目の並び順の指定には使えるのでうまく活用しましょう。

st.Pageとst.navigationを使う方法

もう一つの方法は、専用のメソッド、st.Pageとst.navigationを使って明示的に複数ページを定義することです。

この方法であれば、端末内の別の場所に置いてあるファイルを呼び出すこともできますし、分岐等を用いて、リンクを配置するページを出し分けることもできます。

具体的な方法は次のようになります。

# ファイル構成
your-repository/
├── page_1.py
├── page_2.py
└── streamlit_app.py

# streamlit_app.pyの中身
import streamlit as st

pg = st.navigation([st.Page("page_1.py"), st.Page("page_2.py")])
pg.run()

st.Page(“ファイル名”)でファイルを読み込んでページオブジェクトを作り、それをst.navigation に渡して、pg.run() しています。run()は忘れがちなので注意しましょう。

正直、pagesディレクトリを使う方がずっと単純なのですが、大規模になってくるとst.navigation を使う場面も出てくると思うので頭の片隅にでもおいておいてください。

今回もStreamlitの記事です。ページレイアウトをスッキリさせるのに便利なテクニックを2つ紹介します。

これまでの記事でも書いてきた通り、Strreamlitは原則的に上から下へと要素が足され続けていきます。(サイドバーへの配置や、プレースホルダーを使って上に遡って要素を配置する等の例外はありましたが。)

ただ、メインカラムに配置したい要素が多くなってくるとページがどうしても縦長になってしまって不便なことがあります。このような時に、サイドバーではなくメインカラム内を複数列に切って要素を配置したり、タブで表示する要素を切り替えたりすると便利です。

複数カラムとタブ切り替えは全然違う機能なのですが縦長になるページをスッキリさせたいという同じ目的で使えるのと、コードで実装した時の書き方が似てるのでまとめてこの記事で紹介します。

Streamlitで複数カラムに要素を配置する方法

まず、画面を縦に分割して複数列に要素を配置する方法です。これは、st.columns() を使います。ドキュメントはこちら。
参考: st.columns – Streamlit Docs

st.columns(3) のように、分割したいカラム数(この例では3個)と指定したり、st.columns([1, 2, 3]) のように、配列で数値を渡すと、1:2:3の長さの比率になるように3分割できたりします。

そして、 st.colunms() が指定したカラムの数だけオブジェクトを返してくるので、そいつをwith文で使います。サンプルコードはこちらです。

import streamlit as st


st.title("複数列のサンプル")

col1, col2, col3 = st.columns(3)

with col1:
    st.write("1列目の中身")
    st.write("サンプルテキスト サンプルテキスト サンプルテキスト サンプルテキスト ")

with col2:
    st.write("2列目の中身")
    st.write("サンプルテキスト サンプルテキスト サンプルテキスト サンプルテキスト ")

with col3:
    st.write("3列目の中身")
    st.write("サンプルテキスト サンプルテキスト サンプルテキスト サンプルテキスト ")

st.write("withを抜けると元通り1カラムのレイアウトになる")

実行した見た目はこのような感じ。無事に複数列のレイアウトが実現できていますね。

タブ切り替えの利用

続いてタブ切り替えの利用です。こちらは、選択したタブに対応したコンテンツだけヒョじすることで表示するものを最小化することができます。

先に注意しておきたいのは、選択しなかったタブの中身のコンテンツはブラウザ上表示されないだけで、サーバー側ではコードは実行されているということです。(今日時点では選択したタブの中身のコードだけ実行するといった機能は用意されていません)

使い方はカラムととても似ていますが、一点違うのは、カラム数のようにタブ数を整数で渡すということはできず、タブのタイトルの配列を渡す必要があります。詳しくは、st.tabs()のドキュメントを参照してください。
参考: st.tabs – Streamlit Docs

サンプルコードはこちらです。

import streamlit as st


st.title("タブ切り替えのサンプル")

tab1, tab2, tab3 = st.tabs(["タブ1", "タブ2", "タブ3"])

with tab1:
    st.write("タブ1の中身")

with tab2:
    st.write("タブ2の中身")

with tab3:
    st.write("タブ3の中身")

これでタブ切り替えが実現できます。表示したい画像を切り替えるとか、オプションを隠しておくとか色々使い道がありそうですね。

データ分析においては表示するグラフを切り替える、といった使い道ももちろんあります。
ただ、先ほども書きましたが、これは選択しなかったタブの中身を表示しない、というだけの機能でコードは一通り実行されるものなので、もしデータの取得や計算などをタブの中に書いていたら表示しない分も無駄に実行されますので注意してください。