やりたいこと

Dr.Sumのpython連携機能で、ボストンの不動産の価格を予測してみようと思います。
（Dr.Sumのpython連携機能についての紹介がメインですので、分析手法については触れません）

Dr.Sumというデータベース製品に、python連携機能が追加されたようです。
Dr.Sumサーバ内でpython実行環境を提供しており、SQLベースでpythonを実行します。
Dr.Sumの中に入っているテーブルやビューを入力データとして、機械学習などを行うことができます。

データの準備

まずは、ボストンの不動産データを学習用・テスト用に分けたcsvを作成して、Dr.Sumにインポートします。

data_preparation.py

import numpy as np
import pandas as pd
from sklearn.datasets import load_boston

def main():
  boston = load_boston()
  input = pd.DataFrame(data = boston.data, columns = boston.feature_names).assign(MEDV=np.array(boston.target))
  
  # 訓練データとテストデータに8:2で分割
  split_index = int(input.shape[0]*0.8)
  
  # 訓練データをcsvに
  input_train = input[:split_index]
  input_train.to_csv("boston_train.csv")
  
  # 実際は、予測対象であるテストデータには教師データ（答えのカラム=MEDV）は存在しないので削除
  input_test = input[split_index:].drop("MEDV", axis = 1)
  input_test.to_csv("boston_test.csv")

if __name__ == "__main__":
  main()

出力したcsvをDr.Sumにインポートします。こちらの記事が参考になると思います。

今回はPython_RFというDBの中に、boston_trainとboston_testというテーブルを作成しました。

インメモリサーバーを使ってインメモリ化する必要があるので、それぞれインメモリ化しときましょう。
インメモリ化の方法は以下です。
・「Dr.Sum 5.6 In-Memory Server」のサービスが立ち上がっているか確認
ついでに、Launch Serverも立ち上がっているのも確認。python連携機能を使うために必要です。

・テーブルの上で右クリック→「プロパティ」→「インメモリ化する」にチェック→OKでプロパティは閉じる。
・もう一度テーブルの上で右クリック→「インメモリ」→「データをロード」

学習

ライブラリのインストール

必要なライブラリをDr.Sumのpython環境にインストールします。
ローカルのpython環境とは、別の環境なので注意です。

C:\DrSum56\AdminTools\udtf-python\python\python.exe -m pip install scikit-learn
C:\DrSum56\AdminTools\udtf-python\python\python.exe -m pip install pandas
C:\DrSum56\AdminTools\udtf-python\python\python.exe -m pip install joblib

アウトプット

コードの紹介の前に、学習フェーズではどういったアウトプットを出すのかを紹介しておきます。
冒頭に「SQLベースでpythonを実行」と書いた通り、Dr.SumのSQL Executor機能を使って画像のようにSQLを実行します。

SELECT
*
FROM
udtf::serial_py(
    boston_train,
    py_file_path='PY_SCRIPT_ROOT/train.py',
    func_name='train',
    schema_func_name='model_status'
) T

テーブル名（5行目）、スクリプトのパス（6行目）、処理関数（7行目）、結果表スキーマ取得関数（8行目）を引数に取ります。
実行するとこんな感じ↓でアウトプットが出てきます。

アウトプットには各説明変数の重要度と、root mean-squared error、訓練データとバリデーションデータの正解率を出しています。
RM（＝1住戸あたりの平均部屋数）の重要度が約63%と価格の予測に大きく影響することや、訓練データで約98%、バリデーションデータで約76%の予測ができていることが分かりました。
こういった形でモデルの評価を行うことができますね。

スクリプト

ではスクリプトの紹介に入ります。

train.py

import io
import pandas as pd
from sklearn.model_selection import train_test_split
import joblib

# 結果表スキーマ取得関数
def model_status(input_schema):
    return ['description VARCHAR', 'value NUMERIC']

# 処理関数
def train(context, input):
    # 訓練データとバリデーション用データを8:2で分割
    X_train, X_val, y_train, y_val = train_test_split(
        input.drop("MEDV", axis = 1).values, 
        input["MEDV"].values, 
        test_size=0.2, random_state=0
    )
    from sklearn.ensemble import RandomForestRegressor
    from sklearn.metrics import mean_squared_error
    
    model = RandomForestRegressor(100, random_state = 0) # ランダムフォレスト回帰モデル
    model.fit(X_train, y_train) # 学習
    y_pred = model.predict(X_val) # 予測
    mse= mean_squared_error(y_val, y_pred) # 評価
    
    # アウトプットとなるdataframeを定義
    model_status = pd.DataFrame(columns = ["description", "value"])
    # 各説明変数の重要度データをアウトプットとして追加する。
    for i, feat in enumerate(input.columns[input.columns!="MEDV"]):
        model_status.loc[feat] = [feat, model.feature_importances_[i]]
    
    #　root mean-squared errorと、訓練データでの正解率及びバリデーション用のデータの正解率をアウトプットに追加
    model_status.loc["RMSE"] = ["RMSE", mse** 0.5]
    model_status.loc["train data"] = ["train data", model.score(X_train, y_train)]
    model_status.loc["validation data"] = ["validation data", model.score(X_val, y_val)]
    
    # モデルの保存
    stream = io.BytesIO()
    joblib.dump(model, stream)
    context.write_remote_file(f"PY_DATA_ROOT/model_RF.sav", stream.getvalue())
  
    return model_status

あれ？メイン関数がないやん！と思うかもしれないですが、先程紹介したSQL文の通りスクリプトに必要なものは処理関数と結果表スキーマ取得関数です。
このtrain.pyを PY_SCRIPT_ROOT/（デフォルトではC:/DrSum56/Server/samples/udtf-py/script）の下に配置するだけでOKです。
とはいえ、実際にはエミュレータを使ってデバッグしながらスクリプトを書いていく（＋デバッグが通って初めて結果表スキーマ取得関数が得られる）ので、記事の最後に紹介します。
処理関数（今回はtrain関数）に関しては、モデルの保存部分以外はDr.Sumのpython環境でなくてもデバッグすることは可能です。
今は一旦、こちらを PY_SCRIPT_ROOT/の下に配置して、SQL Executorで実行してみてください。
モデルの学習、評価、保存まで行います。
モデルはPY_DATA_ROOT/（デフォルトではC:/DrSum56/Server/samples/udtf-py/data）の下にmodel_RF.savという名前で保存されています。

処理関数（今回はtrain関数）のシグネチャについて

context: Context・・・モデルの保存やロードを行うためのオブジェクトです。
input: pandas.DataFrame・・・入力データです。Pandas.DataFrame型です。
return Optional[pandas.DataFrame]・・・出力データです。Pandas.DataFrame型です。
マニュアルページはこちら

予測

スクリプト

モデルの学習ができたので、予測フェーズに入ります。
スクリプトでは、モデルのロードと予測を行っています。
学習のスクリプトよりはスッキリしてると思います。
学習の時と同様にPY_SCRIPT_ROOT/（デフォルトではC:/DrSum56/Server/samples/udtf-py/script）の下にpredict.pyを配置してください。

predict.py

import pandas as pd
import joblib, io

def predict_schema(input_schema):
    return ['CRIM NUMERIC', 'ZN NUMERIC', 'INDUS NUMERIC', 'CHAS NUMERIC', 'NOX NUMERIC', 'RM NUMERIC', 'AGE NUMERIC', 'DIS NUMERIC', 'RAD NUMERIC', 'TAX NUMERIC', 'PTRATIO NUMERIC', 'B NUMERIC', 'LSTAT NUMERIC', 'MEDV NUMERIC']

def predict(context, input):
    load_stream = io.BytesIO(context.read_remote_file(f"PY_DATA_ROOT/model_RF.sav"))
    model = joblib.load(load_stream)
    y_pred = model.predict(input.values) # 予測
    output = input.assign(MEDV = y_pred)
    
    return output

アウトプット

学習と同様に下のSQL文をSQL Executorで実行してみてください。

SELECT
*
FROM
udtf::serial_py(
    boston_test,
    py_file_path='PY_SCRIPT_ROOT/predict.py',
    func_name='predict',
    schema_func_name='predict_schema'
) T

アウトプットのテーブルとして、テストデータと予測した価格（=MEDV）が表示されます。

まとめ

Dr.Sumのpython連携機能を使って不動産価格の予測を行いました。
データサイエンティストの方はODBCやJDBCでDr.Sumとの接続してモデルを作成し、エンドユーザは保存されたモデルを使って予測値を見るといったような運用になるのかなぁ～と想像してます。

【Appendix】エミュレータを使ったデバッグ

メインの方では省略しましたが、実際はエミュレータを使ったデバッグを行いながらスクリプトを書いていきます。
デバッグの際はDr.Sumのpython環境独自に存在するdspyライブラリを使います。

Dr.Sumからのデータのインポートは今回pyodbcライブラリを使ったODBC接続なので、インストールを行います。

C:\DrSum56\AdminTools\udtf-python\python\python.exe -m pip install pyodbc

train.py

import io
import pandas as pd
from sklearn.model_selection import train_test_split
import joblib

# Dr.SumからODBC接続で訓練データを取得
def import_data():
  import pyodbc
  con = pyodbc.connect(DRIVER="Dr.Sum 5.6 ODBC Driver",
            HOST="[Dr.SumサーバのIP]", DATABASE="Python_RF", trusted_connection="yes",
            user="[ID]", password="[PW]", port = "6001"
            )
  TBL = "boston_train"
  query = "SELECT * FROM {tbl} ;".format(tbl=TBL)
  train_data = pd.read_sql(query, con)
  return train_data

def train(context, input):
  # 訓練データとバリデーション用データを8:2で分割
  X_train, X_val, y_train, y_val = train_test_split(
      input.drop("MEDV", axis = 1).values, 
      input["MEDV"].values, 
      test_size=0.2, random_state=0
  )
  from sklearn.ensemble import RandomForestRegressor
  from sklearn.metrics import mean_squared_error
  
  model = RandomForestRegressor(100, random_state = 0) # ランダムフォレスト回帰モデル
  model.fit(X_train, y_train) # 学習
  y_pred = model.predict(X_val) # 予測
  mse= mean_squared_error(y_val, y_pred) # 評価
  
  # アウトプットとなるdataframeを定義
  model_status = pd.DataFrame(columns = ["description", "value"])
  # 各説明変数の重要度データをアウトプットとして追加する。
  for i, feat in enumerate(input.columns[input.columns!="MEDV"]):
      model_status.loc[feat] = [feat, model.feature_importances_[i]]
  
  #　root mean-squared errorと、訓練データでの正解率及びバリデーション用のデータの正解率をアウトプットに追加
  model_status.loc["RMSE"] = ["RMSE", mse** 0.5]
  model_status.loc["train data"] = ["train data", model.score(X_train, y_train)]
  model_status.loc["validation data"] = ["validation data", model.score(X_val, y_val)]
  
  # モデルの保存
  stream = io.BytesIO()
  joblib.dump(model, stream)
  context.write_remote_file(f"PY_DATA_ROOT/model_RF.sav", stream.getvalue())
  
  return model_status

def main():
  import dspy.emulator
  train_data = import_data()
  
  # エミュレータを使ったデバッグ
  builder=dspy.emulator.EmulatorBuilder(train_data)
  emulator=builder.buildSerial(
      py_file_path = "PY_SCRIPT_ROOT/train.py",
      func_name = "train",
      schema_func_name = "model_status",
  )
  result=emulator.execute(create_schema_mode = True) # Trueで結果表スキーマ取得関数を出力するようにする
  print(result)

if __name__ == "__main__":
  main()

入力がcsvからだった場合、Dr.Sumが意図している型と異なる場合があるので、エミュレータを使って型を修正したりできます。
エミュレータを使う理由についてはマニュアルを参照してください。

スクリプトを実行する際は、Dr.Sumサーバ内のpython環境を使ってください。

C:\DrSum56\AdminTools\udtf-python\python\python.exe C:\[デバッグを行っているディレクトリ]\train.py

デバッグが成功すると下記のようなメッセージがコンソールに出てきます。

以下のスタブ関数をC:\[デバッグを行っているディレクトリ]\train.pyにコピーしてください。

--------------------------------------------------------------------------------
def model_status(input_schema):
    return ['description VARCHAR', 'value NUMERIC']
--------------------------------------------------------------------------------

メッセージに従って、結果表スキーマ取得関数をスクリプトに追記します。
私のVScode環境ではショートカットでローカルのpython環境で実行するようにしているので、正直なところ毎回Dr.Sumのpython.exeのパスを指定するのが手間ですね。
処理関数の引数と戻り値がpandas.DataFrame型であることを意識しておけば、学習の部分等はローカルの環境でデバッグしたほうがいい気がします。

参考

・1から進める場合はマニュアルのチュートリアルを進めても良いかと思います。

・python連携に関するQiita記事もありました。