PySparkで線形回帰モデルを作成する

# import
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression

# VIFを格納するリスト
VIF_list = []

# 説明変数の取得
tmp_ex_var = z_df.columns
ex_var = [ev for ev in tmp_ex_var if "_z_scaled" in ev] # 標準化した変数のみ取得
ex_var.remove("target_z_scaled") # 目的変数を排除

# VIFの計算
for c in ex_var:

  # featuresの作成

  ##VIF計算時に説明変数となる変数の取得
  VIF_ex_var = [c2 for c2 in ex_var if c2 != c]

  ## VectorAssemblerインスタンス作成
  vector_assembler = VectorAssembler(
      inputCols = VIF_ex_var,
      outputCol='features'
  )

　## 線形回帰モデルのインプットとなるpyspark.DataFrameの作成
  VIF_z_df = (
      vector_assembler
      .transform(z_df)
      .withColumnRenamed(c, "label")
  )

  # 線形回帰
  lr = LinearRegression(regParam = 0.0, fitIntercept = False)
  lrModel = lr.fit(VIF_z_df)
  trainig_summary = lr.summary

  ## VIFの計算
  VIF_list.append(1/(1-trainig_summary.r2))

　## 今何を計算しているか可視化
  print("目的変数:", c, "\n")
  print("説明変数:", VIF_ex_var, "\n")

# pandasDataFrameに格納してVIFを見やすくする
VIF_df = pd.DataFrame(index = ex_var)
VIF_df["VIF"] = VIF_list 

# import
from pyspark.ml.feature import VectorAssembler

# featuresの作成
vector_assembler = VectorAssembler(
      inputCols = ex_var, # featuresに入れたいカラムをリストで指定
      outputCol = 'features'
)

feature_z_df = (
      vector_assembler
      .transform(z_df) # featuresが含まれたpyspark.DataFrameの作成
      .withColumnRenamed(target_var, "label") # LinearRegressionで指定する目的変数のデフォルトがlabelなのでlabelにカラム名を変えた
)

# 線形回帰モデルの作成
lr = LinearRegression(regParam = 0.0, fitIntercept = True) # 定数項がある場合
# lr = LinearRegression(regParam = 0.0, fitIntercept = False) # 定数項がない場合

lrModel = lr_intercept.fit(feature_z_df)

# 予測値の計算
predict_summary = lrModel.evaluate(feature_z_df)

# Pandasに変換
sampled_predict_residuals_pd_df = (
    predict_residuals_df
    .select("prediction", "residuals")
    .toPandas() #Pandasに変換
)

# 散布図(予測値×残差)作成
import matplotlib.pyplot as plt
plt.scatter(sampled_predict_residuals_pd_df["prediction"], sampled_predict_residuals_pd_df["residuals"])
plt.axhline(y = 0, xmin = min(sampled_predict_residuals_pd_df["prediction"]), xmax = max(sampled_predict_residuals_pd_df["prediction"]), color = "black", linestyle = "dotted")

# ランダムサンプリングする割合
rate = 0.1

# ランダムサンプリング & Pandasに変換
sampled_predict_residuals_pd_df = (
    predict_residuals_df
    .select("prediction", "residuals")
    .sample(rate) # ランダムサンプリングしている箇所はココ！
    .toPandas() #Pandasに変換
)

# QQプロット作成
import scipy.stats as stats
stats.probplot(sampled_predict_residuals_pd_df["residuals"], dist = "norm", plot = plt)
plt.show()