PythonとSpark Machine Learningを使用した分類

ポイント: Pythonを使用したSpark MLは、心臓欠陥などの健康状態を予測する分類モデルの開発を促進します。このチュートリアルは、データの準備、変換、およびロジスティック回帰の実装方法を示しています。

SparkのDataFrameを活用し、Pythonライブラリで大規模なデータセットを効率的に処理します。
完全なワークフローを通じて、93%のモデル精度を達成します。

私たちは、Scalaプログラミング言語を使用してSpark MLを使用する方法について書いてきました。しかし、多くのプログラマーはScalaを知りません。Pythonは機械学習の強さに基づいて、ユーザー数においてJavaを追い越しました。そこで、Spark MLをPythonで使用することに焦点を当てましょう。

SparkはScala中心といえるでしょう。ScalaにはPythonとScalaのインターフェースとコマンドラインインタープリターがあります。Scalaがデフォルトです。Pythonのものはpysparkと呼ばれます。Sparkが提供する例のほとんどはScalaであり、場合によってはPythonの例は提供されていません。

Apache Atom

Pythonは、NumPy、Pandas、matplotlibのため、データサイエンスに最適な言語です。これらのツールは、配列を扱い、チャートを描くのを容易にし、大規模なデータ配列を効率的に処理できます。しかし、Sparkはクラスター全体に広がる巨大なデータを扱うために設計されています。両方のツールを使用し、需要に応じて行き来することが良い習慣です。

しかし、私たちが見るように、SparkのデータフレームはPandasのデータフレームとは異なるため、これらのすべてのオブジェクト間に100%の互換性はありません。matplotlibでプロットするために、Sparkのデータフレームをリストや配列、その他の構造に変換する必要があります。馴染みのある「[:,4]」で配列をスライスできないため、同じ操作を行うにはより多くのコードが必要です。

もう一つの問題はパフォーマンスです。Apache Atomは、JavaプロセスのオブジェクトをPythonプロセスに効率的に変換するために存在します。SparkはJavaとScalaで書かれています。ScalaはJavaの上に乗っています。もちろん、PythonはPythonプロセスで動作します。

Arrowは、SparkのデータフレームをPandasのデータフレームに変換し、.strong>withcolumn()などの列単位の操作を行う際の処理を高速化します。

Sparkは、このパフォーマンスの向上を利用するためにSparkで行う必要がある設定変更とその周辺の問題について説明しています。Apache Arrowのドキュメントを参照してください。

心臓病患者データ

データをサンパウロ大学のデータセットからダウンロードします。利用可能なこちら。興味があれば、この議論をご覧ください。

列は次のとおりです:

年齢
性別
胸痛のタイプ（4つの値）
安静時血圧
血清コレステロール（mg/dl）
空腹時血糖 > 120 mg/dl
安静時心電図結果（値 0,1,2）
達成した最大心拍数
運動誘発性狭心症
Oldpeak = 安静時に対する運動によって誘発されたST抑圧
ピーク運動STセグメントの傾斜
フルオロスコピーで染色された主要な血管の数（0-3）
Thal: 3 = 正常; 6 = 固定欠陥; 7 = 可逆的欠陥

コードの説明

目標は、Spark MLとPythonを使用して、誰かが心臓欠陥を持っているかどうかを予測するバイナリロジスティック回帰モデルを構築することです。以下のコードは、Zeppelinノートブックで利用可能です。こちら。

最初に、データを読み込み、列名を割り当てます。データが小さいため、Pandasが簡単であるため、Pandasのデータフレームに読み込みます。その後、spark.createDataFrame()を使用してSparkのデータフレームに変換します。

SparkのデータフレームがPandasのいくつかの機能を欠いていることを理解できるかもしれません。特に、Pandasを使用することで、最初の13列を取得し、最後の列を削除するために.iloc()を使用できるからです。この最後の列はデータに対して意図されていないノイズのようです。

    %spark.pyspark

import pandas as pd
from pyspark.sql.types import StructType, StructField, NumericType

 
cols = ('age',       
      'sex',         
       'chest pain',           
       'resting blood pressure',    
       'serum cholesterol',       
       'fasting blood sugar',         
       'resting electrocardiographic results', 
       'maximum heart rate achieved',  
       'exercise induced angina',     
       'ST depression induced by exercise relative to rest',  
      'the slope of the peak exercise ST segment',     
      'number of major vessels ',       
       'thal',  
       'last')
      

      
data = pd.read_csv('/home/ubuntu/Downloads/heart.csv', delimiter=' ', names=cols)

data = data.iloc[:,0:13]

data['isSick'] = data['thal'].apply(isSick)

df = spark.createDataFrame(data)

フィールドthal は、患者が心臓の問題を持っているかどうかを示します。数値は次のとおりです:

値3は患者が健康であることを意味します（正常）。
値6は患者の健康問題が解決されたことを意味します。
値7は患者の健康問題が解決可能であることを意味します。

ですから、この関数isSick()を書いて、0を陰性、1を陽性としてフラグ付けします。バイナリロジスティック回帰は2つの結果のいずれかを必要とします。

    def isSick(x):
    if x in (3,7):
        return 0
    else:
        return 1

機械学習と分類または回帰の問題では:

患者の年齢や血糖値などの特徴の行列
患者が心臓の問題を持っているかどうかを示すラベルのベクトル

私たちはZeppelinノートブックを使用しているため、SparkのPythonコマンドシェルであるPySparkを使用するために、各Zeppelinセルの最上部に%spark.pysparkと記述します。

次に、dfデータフレームのどの列を特徴として使用するかを示します。そして、VectorAssemblerを使用して、これらの12のフィールドをすべて含む新しい列をfeaturesと呼ばれる配列として作成します。

次に、transform()操作を使用して、raw_dataというSparkデータフレームを作成し、特徴列のみを選択します。

    %spark.pyspark
from pyspark.ml.feature import StandardScaler
from pyspark.ml.feature import VectorAssembler

features =   ('age',       
      'sex',         
       'chest pain',           
       'resting blood pressure',    
       'serum cholestoral',       
       'fasting blood sugar',         
       'resting electrocardiographic results', 
       'maximum heart rate achieved',  
       'exercise induced angina',     
       'ST depression induced by exercise relative to rest',  
      'the slope of the peak exercise ST segment',     
      'number of major vessels ') 

assembler = VectorAssembler(inputCols=features,outputCol="features")
 
raw_data=assembler.transform(df)
raw_data.select("features").show(truncate=False)

私たちはStandard Scalerを使用して、すべての数値を同じスケールに揃えます。これは機械学習の標準的な慣行です。これは観測値から平均を引き、その後標準偏差で割ります。

    %spark.pyspark
from pyspark.ml.feature import StandardScaler

standardscaler=StandardScaler().setInputCol("features").setOutputCol("Scaled_features")
raw_data=standardscaler.fit(raw_data).transform(raw_data)
raw_data.select("features","Scaled_features").show(5)

ここに特徴データが現在どのように見えるかがあります:

    +--------------------------------------------------------+
|features                                                |
+--------------------------------------------------------+
|[70.0,1.0,4.0,130.0,322.0,0.0,2.0,109.0,0.0,2.4,2.0,3.0]|
|[67.0,0.0,3.0,115.0,564.0,0.0,2.0,160.0,0.0,1.6,2.0,0.0]|
|[57.0,1.0,2.0,124.0,261.0,0.0,0.0,141.0,0.0,0.3,1.0,0.0]|
|[64.0,1.0,4.0,128.0,263.0,0.0,0.0,105.0,1.0,0.2,2.0,1.0]|
|[74.0,0.0,2.0,120.0,269.0,0.0,2.0,121.0,1.0,0.2,1.0,1.0]|
|[65.0,1.0,4.0,120.0,177.0,0.0,0.0,140.0,0.0,0.4,1.0,0.0]|
|[56.0,1.0,3.0,130.0,256.0,1.0,2.0,142.0,1.0,0.6,2.0,1.0]|
|[59.0,1.0,4.0,110.0,239.0,0.0,2.0,142.0,1.0,1.2,2.0,1.0]|
|[60.0,1.0,4.0,140.0,293.0,0.0,2.0,170.0,0.0,1.2,2.0,2.0]|
|[63.0,0.0,4.0,150.0,407.0,0.0,2.0,154.0,0.0,4.0,2.0,3.0]|
|[59.0,1.0,4.0,135.0,234.0,0.0,0.0,161.0,0.0,0.5,2.0,0.0]|
|[53.0,1.0,4.0,142.0,226.0,0.0,2.0,111.0,1.0,0.0,1.0,0.0]|

通常通り、データをトレーニングとテストのデータセットに分割します。データが少ないため、50/50の分割を使用します。

    %spark.pyspark
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit

training, test = raw_data.randomSplit([0.5, 0.5], seed=12345)

次に、ロジスティック回帰モデルを作成し、トレーニングします。つまり、モデルがラベル列isSickにある結果に最も近い係数と切片を計算させます。

    %spark.pyspark
from pyspark.ml.classification import LogisticRegression

lr = LogisticRegression(labelCol="isSick", featuresCol="Scaled_features",maxIter=10)
model=lr.fit(training)
predict_train=model.transform(training)
predict_test=model.transform(test)
predict_test.select("isSick","prediction").show(10)

ここで、並べて比較して最初の数行を示します。これらは、ほとんどの場合、正しいです。

    +------+----------+
|isSick|prediction|
+------+----------+
|     0|       0.0|
|     1|       0.0|
|     0|       0.0|
|     0|       0.0|
|     0|       0.0|
|     0|       0.0|
|     0|       0.0|
|     0|       1.0|
|     0|       1.0|
|     0|       0.0|
+------+----------+

これが係数と切片を示しています。

ここにあります:

    %spark.pyspark
print("Multinomial coefficients: " + str(model.coefficientMatrix))
print("Multinomial intercepts: " + str(model.interceptVector))

ここにあります:

    Multinomial coefficients: DenseMatrix([[-0.41550466,  1.21573123,  0.16600826,  0.36478609,  0.33716549,
              -0.020686  , -0.2092542 , -0.86514924,  0.1427418 , -0.3610241 ,
               0.57324392,  0.42563706]])
Multinomial intercepts: [-0.2767309778166021]

今、私たちはいくつかのSpark SQL関数Fを使用して、IsSickとpredictionが等しいときに新しい列correctを作成します。これは、予測結果が実際の結果に等しいことを意味します。

    %spark.pyspark
import pyspark.sql.functions as F
check = predict_test.withColumn('correct', F.when(F.col('isSick') == F.col('prediction'), 1).otherwise(0))
check.groupby("correct").count().show()

ここに結果があります:

    +-------+-----+
|correct|count|
+-------+-----+
|      1|  137|
|      0|   10|
+-------+-----+

したがって、精度は137 / 137 + 10 = 93%

モデルの精度を示す他の方法もありますが、曲線の下の面積のようなものです。しかし、これは理解しやすい最も簡単な方法です。経験豊富なデータサイエンティストや統計学者でない限り、私たちは将来の記事で精度を確認するより複雑な方法を説明します。

Apache Sparkガイド

これらの投稿は私自身のものであり、必ずしもBMCの立場、戦略、または意見を表すものではありません。

ワークフローオーケストレーション

メインフレーム簡素化

人気の目的地

PythonとSpark Machine Learningを使用した分類

Apache Atom

心臓病患者データ

コードの説明

Apache Sparkガイド

メインフレーム用に構築されたAIエージェント

人工知能（AI）の関連記事

The Briefを購読する

ありがとう

ご要望またはご協力できることがございましたらお気軽にお問い合わせください

セールス & お見積り

ヘルプ & サポート

人気の目的地

PythonとSpark Machine Learningを使用した分類

Apache Atom

心臓病患者データ

コードの説明

Apache Sparkガイド

人工知能（AI）の関連記事

The Briefを購読する

ありがとう