データサイエンスの文脈で理解するデザインパターン
はじめに
エンジニアではなくデータサイエンティストとしてデザインパターンについて初めて勉強したとき、なんとなくは理解しつつも正直なところ実務での利用イメージがピンと来ませんでした。その理由の一つとして、データサイエンスの現場では特に、教科書的なデザインパターンの例がそっくりそのまま使われることは少ない、あるいはそれに気づかないことが多いためだと思われます。
そこで本記事では、データサイエンスのフレームワークやその使用例を通してそれぞれのパターンを見ることで、デザインパターンをより身近なものに感じられるようにすることを試みます。
そもそもデザインパターンとは
デザインパターンとは、特定の文脈で繰り返し現れる設計問題に対する再利用可能な解決策のことです。GoFによる分類では、以下の3カテゴリに整理されます。
- 生成に関するパターン：オブジェクトの生成方法を管理するパターン
- 構造に関するパターン：複数のオブジェクト間の関係や構造を定義するパターン
- 振る舞いに関するパターン：オブジェクトの動作や振る舞いを定義するパターン
デザインパターンが提案された当時と比較して、現代では洗練されたAPIを持つフレームワークやライブラリが多く提供されているため、すべてのパターンが有用とは限りません。その一方で、デザインパターンを学ぶことには以下のメリットがあるため、依然として学ぶ価値はあるものだと思われます。
-
共通言語としての利用
開発者間で会話する上での前提知識となり、設計方針に関する議論をスムーズに進めることができます。 -
品質の高い設計方法の学習
多くのパターンはオブジェクト指向におけるSOLID原則と深い関係にあるため、デザインパターンを理解することを通して、保守性・拡張性の高い実装方針を学ぶことができます。
デザインパターンの例
以下ではデータサイエンスの文脈と関係が深いパターンをピックアップして紹介します。なお、それぞれのコード例は説明用に簡略化しています。
生成に関するパターン
Factory Method
インスタンス生成処理を呼び出し側から隠蔽するパターンです。これにより、呼び出し側のコードを変更することなく新しい型を追加することができます。以下は異なる機械学習モデルのファクトリークラスを定義し、統一されたインターフェースで呼び出す例です。
from abc import ABC, abstractmethod
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
class ModelFactory(ABC):
@abstractmethod
def create_model(self):
pass
class LogisticRegressionFactory(ModelFactory):
def create_model(self):
return LogisticRegression()
class RandomForestFactory(ModelFactory):
def create_model(self):
return RandomForestClassifier()
def get_model(factory: ModelFactory):
return factory.create_model()
factory = LogisticRegressionFactory()
model = get_model(factory) # モデルの種類を意識せず get_model でインスタンス化
本来のFactory Methodはクラスベースですが、Pythonでは関数を使うことでより軽量に実装することも可能です。以下の例ではMODEL_MAPに新しいモデルクラスを登録することで、容易に新しいモデルを追加・登録することができます。
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
MODEL_MAP = {"logistic": LogisticRegression, "rf": RandomForestClassifier}
def create_model(model_type: str):
if model_type not in MODEL_MAP:
raise ValueError(f"Unknown model type: {model_type}")
return MODEL_MAP[model_type]()
model = create_model("logistic")
Abstract Factory
Factory Methodは1種類のインスタンスを生成するためのパターンですが、複数の関連するオブジェクト群をセットで生成したい場合はAbstract Factoryが使えます。これにより、オブジェクトの組合せの整合性を保証することができます。例えば、モデルごとに対応する前処理が異なる場合、モデルと前処理インスタンスの組み合わせをAbstract Factoryで生成します。
from abc import ABC, abstractmethod
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import FunctionTransformer, StandardScaler
class AbstractFactory(ABC):
@abstractmethod
def create_preprocessor(self):
pass
@abstractmethod
def create_model(self):
pass
class LogisticRegressionFactory(AbstractFactory):
def create_preprocessor(self):
return StandardScaler()
def create_model(self):
return LogisticRegression()
class RandomForestFactory(AbstractFactory):
def create_preprocessor(self):
return FunctionTransformer()
def create_model(self):
return RandomForestClassifier()
def setup_pipeline(factory: AbstractFactory):
preprocessor = factory.create_preprocessor()
model = factory.create_model()
return preprocessor, model
factory = LogisticRegressionFactory()
preprocessor, model = setup_pipeline(factory) # 前処理とモデルの組をまとめてインスタンス化
構造に関するパターン
Adapter
異なるインターフェースを持つ機能同士を、統一したインターフェース上で扱えるようにするパターンです。例えば、一般的な機械学習モデルをsklearn準拠のインターフェースに統一するケースが考えられます。これにより、sklearn組み込みで提供されているモデルと同じように、pipelineへの組み込みなどが行えるようになります。
import numpy as np
import statsmodels.api as sm
from sklearn.base import BaseEstimator, ClassifierMixin
class StatsmodelsLogitAdapter(BaseEstimator, ClassifierMixin): # sklearnのインターフェースを継承
def __init__(self, **sm_kwargs):
self.sm_kwargs = sm_kwargs
self.result_ = None
self.classes_ = np.array([0, 1])
def fit(self, X, y):
X = np.asarray(X)
X = sm.add_constant(X, has_constant="add")
y = np.asarray(y)
model = sm.Logit(y, X, **self.sm_kwargs)
self.result_ = model.fit(disp=False)
return self
Facade
複雑な処理を隠蔽し、シンプルなインターフェースを提供するパターンです。例えば、複雑な工程を踏むデータの前処理をシンプルなインターフェースで提供するケースが挙げられます。
from pathlib import Path
def _load_and_validate(path):
...
def _clean_data(df):
...
def _feature_engineering(df):
...
def _encode_and_split(df):
...
def prepare_data(path: Path): # シンプルな関数に複雑な処理を集約
df = _load_and_validate(path)
df = _clean_data(df)
df = _feature_engineering(df)
X, y = _encode_and_split(df)
return X, y
Composite
「個」と「集合」を同じインターフェースで扱えるようにするパターンです。sklearnのpipelineが例として挙げられます。sklearnのpipelineでは、それぞれの前処理、それらをつなげたパイプラインのいずれもfitやtransformなどの統一したインターフェースを通して扱うことができます。
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import PCA, StandardScaler
scaler = StandardScaler()
scaler.fit(X)
pipeline = Pipeline([("scaler", StandardScaler()), ("pca", PCA(n_components=2))])
pipeline.fit(X)
Decorator
既存の機能をラップする形で新しい機能を持つオブジェクトを生成するパターンです。Pythonではデコレーターが組み込みでサポートされているため頻出パターンです。以下は学習や前処理など、処理にかかった時間をロギングする機能を与えるデコレーターの例です。
import logging
from functools import wraps
from time import perf_counter
logger = logging.getLogger(__name__)
def time_logger(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = perf_counter()
try:
return func(*args, **kwargs)
finally:
end = perf_counter()
logger.info("%s took %.4fs", func.__name__, end - start)
return wrapper
@time_logger
def train():
...
Proxy
あるオブジェクトの代理となるオブジェクトを提供し、代理オブジェクトを通して元のオブジェクトを操作するパターンです。必要になったときまで計算を行わない遅延評価がその一例で、PolarsのLazyFrameがあてはまります。
import polars as pl
lazy_df = pl.scan_csv("large_file.csv") # Proxyを生成
lazy_df = lazy_df.filter(pl.col("age") > 30).select(["name", "age"])
result = lazy_df.collect() # 遅延評価
振る舞いに関するパターン
Template Method
抽象クラスで枠組みを定義し、具象クラスでその詳細を実装するパターンです。学習ループの枠組みは固定しつつ、具体的な計算ステップを差し替えるPyTorch LightningのLightningModuleが一例です。
import lightning as L
import torch
from torch import nn
class SimpleClassifier(L.LightningModule):
def __init__(self):
super().__init__()
self.model = nn.Sequential(nn.Linear(4, 16), nn.ReLU(), nn.Linear(16, 3))
self.loss_fn = nn.CrossEntropyLoss()
def training_step(self, batch, batch_idx): # メソッドを上書き
x, y = batch
logits = self.model(x)
loss = self.loss_fn(logits, y)
self.log("train_loss", loss)
return loss
def configure_optimizers(self):
return torch.optim.Adam(self.parameters(), lr=1e-3)
Strategy
アルゴリズムを同一のインターフェースに沿って実装することで交換可能にし、柔軟に切り替えることができるようにするパターンです。ML文脈だと登場頻度が高く、前処理やモデルのインターフェースを統一することで、呼び出し側が詳細を気にすることなく様々なアルゴリズムを入れ替えることができます。以下はモデルを入れ替えて性能を評価する例です。
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
STRATEGIES = {
"logistic": LogisticRegression,
"random_forest": RandomForestClassifier,
}
def evaluate(strategy_name, X_train, y_train, X_test, y_test):
model = STRATEGIES[strategy_name]()
model.fit(X_train, y_train)
return accuracy_score(y_test, model.predict(X_test))
State
オブジェクトの内部状態に応じて挙動を切り替えるパターンです。例えば、PyTorchで学習時と評価時での勾配計算やドロップアウトの振る舞いを切り替える例が挙げられます。
import torch
import torch.nn as nn
model = nn.Sequential(nn.Linear(5, 5), nn.ReLU(), nn.Dropout(p=0.5))
model.train()
model.eval()
Observer
対象の状態を登録されたオブジェクトが監視し、特定の処理が行われたタイミングで反応するパターンです。監視対象と反応の疎結合を実現することができます。PyTorch LightningのCallback機能が例として挙げられます。
import lightning as L
from lightning.pytorch.callbacks import Callback
class Classifier(L.LightningModule):
def __init__(self):
super().__init__()
...
class PrintCallback(Callback):
def on_train_epoch_end(self, trainer, pl_module):
print(f"Epoch {trainer.current_epoch} finished.")
trainer = L.Trainer(max_epochs=3, callbacks=[PrintCallback()]) # Observerを登録
trainer.fit(Classifier(), data)
おわりに
以上、データサイエンスの文脈で用いられるデザインパターンについて身近なコード例を通して紹介しました。ここでは触れなかった他のパターンについても、簡潔にですが以下に列挙しておきます。
-
生成に関するパターン
- Singleton：インスタンスが一つだけ存在することを保証するパターン
- Builder：複雑なオブジェクトの構築プロセスを段階的に行うパターン
- Prototype：既存オブジェクトを複製して新しいインスタンスを作成するパターン
-
構造に関するパターン
- Bridge：実装と抽象を分離するパターン
- Flyweight：同一オブジェクトの共有によりメモリ使用量を削減するパターン
-
振る舞いに関するパターン
- Command：処理を実行可能オブジェクトとして表現するパターン
- Chain of Responsibility：複数の処理候補の中から適切な処理を連鎖的に選択するパターン
- Mediator：複数オブジェクト間の複雑な相互作用を仲介オブジェクトを通して整理するパターン
- Memento：オブジェクトの状態を外部に保存して復元可能にするパターン
- Visitor：データ構造と処理を分離し、新しい処理を既存のクラス群に対して追加できるようにするパターン
実例を通して、以前よりは各パターンに関するイメージが付きました。データサイエンスでは往々にして複雑な処理フローを組み立てることが多いため、デザインパターンを意識しつつ質の高い設計ができるように心がけていきたいと思います。
Discussion