🐵
MLパイプラインツールGokartを使ってパイプラインのこと初め

2025/01/08に公開
機械学習
初心者向け
mlパイプラインツール
tech
 はじめにこんにちは！　IT企業でエンジニアをしている和泉と申します。

普段は、Pythonを使ったアプリケーションの開発などに携わっています。

PyConJP2024で初めて知ったMLパイプラインツールGokartを使ったサンプルプログラムを作成したのでご紹介したいと思います。

 なぜこの記事を書いたのか！機械学習（ML）のプロジェクトを進める中で、データの前処理、モデルの学習、評価といった一連の作業を効率的に管理することが重要です。特に、これらの作業を再現可能でスケーラブルな方法で行うには、MLパイプラインツールが役立ちます。本記事では、初心者でも取り組みやすい「Gokart」を用いて、MLパイプラインの基本を紹介します。

 MLパイプラインツールとはMLパイプラインツールは、機械学習プロジェクトにおけるデータの前処理からモデルの評価までのプロセスを効率的に管理するためのツールです。複数のステップを整理し、再現可能で分離された形で開発することで、プロジェクトの複雑性を軽減します。

 なぜMLパイプラインツールが必要なのか機械学習プロジェクトでは、以下の理由でパイプラインツールが重要です:

プロセスの複雑性: データの取得、前処理、モデルの学習・評価といった多くのステップが必要。

変化への対応: データやモデルが頻繁に変化するため、それに追随できる柔軟性が求められる。

効率的な開発: データやコードの管理を簡略化し、効率的な実験が可能になる。

 Gokartの特徴Gokartは、Pythonで実装された軽量なMLパイプラインツールであり、以下の特徴があります:

再現性の高い開発: データや処理を明確に定義し、再現性を確保。

タスク指向の設計: すべての処理をタスクとして記述し、分離して開発。

軽量でシンプル: 学習コストが低く、初心者でも取り組みやすい。

 サンプルプログラム以下のコードは、Gokartを用いた簡単なMLパイプラインの例です。データの前処理、モデルの学習、評価をパイプライン化しています。
import gokart
import pandas as pd

from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import r2_score

# タスク1: データの読み込み
class LoadData(gokart.TaskOnKart):
    def output(self):
        # データの保存先を指定
        return self.make_target('./iris_data.csv')

    def run(self):
        # Irisデータセットをロード
        iris = load_iris()
        data_array = iris.data  # 特徴量データ
        feature = iris.feature_names  # 特徴量名
        target = iris.target  # ターゲットデータ

        # 特徴量とターゲットをDataFrameに変換
        df_data = pd.DataFrame(data_array, columns=feature)
        df_target = pd.DataFrame(target, columns=['target'])
        df = pd.concat([df_data, df_target], axis=1)  # 特徴量とターゲットを結合
        self.dump(df)  # データを保存

# タスク2: データの前処理
class PreTreatment(gokart.TaskOnKart):
    data_df = gokart.TaskInstanceParameter()  # 入力データタスクを指定

    def output(self):
        # スケーリング済みデータとスケーラーの保存先を指定
        return {
            'scaled_data': self.make_target('./iris_data_scaled.csv'),
            'scaler': self.make_target('./scaler.pkl')
        }

    def run(self):
        df: pd.DataFrame = self.load(self.data_df)  # 入力データを読み込み
        scaler = StandardScaler()  # 標準化用のスケーラーを初期化
        X = df.drop('target', axis=1)  # 特徴量データを抽出
        scaler.fit(X)  # スケーラーをフィット
        scaled_data = scaler.transform(X)  # データをスケーリング

        # スケーリング済みデータとスケーラーを保存
        self.dump(pd.DataFrame(scaled_data, columns=X.columns), 'scaled_data')
        self.dump(scaler, 'scaler')

# タスク3: モデルの学習
class TrainModel(gokart.TaskOnKart):
    data_df = gokart.TaskInstanceParameter()  # 入力データタスクを指定
    pre_treatment = gokart.TaskInstanceParameter()  # 前処理タスクを指定

    def output(self):
        # 学習済みモデルの保存先を指定
        return self.make_target('./model.pkl')

    def run(self):
        df: pd.DataFrame = self.load(self.data_df)  # 入力データを読み込み
        X: pd.DataFrame = self.load(self.pre_treatment)['scaled_data']  # スケーリング済みデータを取得
        y: pd.Series = df['target']  # ターゲットデータを取得

        clf = RandomForestClassifier()  # ランダムフォレストモデルを初期化
        clf.fit(X, y)  # モデルを学習
        self.dump(clf)  # 学習済みモデルを保存

# タスク4: 全体のパイプライン
class MLPipleLine(gokart.TaskOnKart):
    def requires(self):
        # 各タスクを順序立てて指定
        data = LoadData()  # データ読み込みタスク
        pre_treatment = PreTreatment(data_df=data)  # 前処理タスク
        train_model = TrainModel(data_df=data, pre_treatment=pre_treatment)  # モデル学習タスク
        return {'data': data, 'pre_treatment': pre_treatment, 'model': train_model}

    def output(self):
        # 最終結果の保存先を指定
        return self.make_target('./model.csv')

    def run(self):
        # 前処理済みデータと元データを読み込み
        X: pd.DataFrame = self.load('pre_treatment')['scaled_data']
        data: pd.DataFrame = self.load('data')
        target = data['target']  # ターゲットデータを取得

        # 学習済みモデルを読み込み
        model = self.load('model')
        result = model.predict(X)  # データを予測

        # R2スコアを計算
        r2_result = r2_score(target, result)
        self.dump(pd.Series(r2_result, name='R2'))  # R2スコアを保存

# メイン処理
if __name__ == '__main__':
    task = MLPipleLine()  # パイプラインタスクを初期化
    output = gokart.build(task)  # パイプラインを実行
    print(output)  # 実行結果を表示

 実行手順必要なライブラリをインストールします: pip install gokart pandas
上記のコードをPythonファイルに保存し、ターミナルで実行します: python sample.py
実行結果として、処理の結果や評価指標が保存されます。

 まとめGokartは、MLパイプラインツールの中でもシンプルで扱いやすいツールです。初心者に最適でありながら、柔軟性や再現性を備えているため、スモールスケールからプロダクションレベルまで幅広く利用可能です。ぜひこの機会にGokartを試してみてください！

なお、今回のソースコードは、こちらのGitHubリポジトリに挙げております。。
はじめに

なぜこの記事を書いたのか！

MLパイプラインツールとは

なぜMLパイプラインツールが必要なのか

Gokartの特徴

サンプルプログラム

実行手順

まとめ

Discussion
