この項ではKaggle Titanic生存者予測をSparkMLというライブラリーを使って実施していきます。
SparkMLとは
Apache Spark の機械学習ライブラリで、シンプルでスケーラビリティが高く、他のツールと容易に統合できるように設計された、機械学習を実装するためのツールです。Spark上で扱う事の出来るパッケージであるため、ビッグデータに対して高速に機械学習処理を行うことが出来るのが特徴です。
Sparkの機械学習ライブラリーとして、SparkMLの他にもSparkMLlibが挙げられます。
SparkMLlibとSparkMLの違いとしては、SparkMLlibはRDDで扱うことを想定して実装されており、SparkMLはDataFrameおよびDatasetで扱うことを想定して開発されています。
チュートリアル
- 機械学習入門者向け ランダムフォレストによる Kaggle Titanic生存者予測を参考にして train.csv, test.csvをダウンロードして、zeppelin/data/ ディレクトリ下の任意の場所に移動します
1. 前処理
1.1 ライブラリーのimport
今回の処理に必要なライブラリーをimportします。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
import org.apache.spark.sql.SparkSession import org.apache.spark.ml._ import org.apache.spark.ml.feature._ import org.apache.spark.ml.classification.RandomForestClassifier import org.apache.spark.sql.functions._ import org.apache.spark.sql.SaveMode import org.apache.spark.sql.DataFrame import org.apache.spark.sql.types.IntegerType import org.apache.spark.ml.tuning.ParamGridBuilder import org.apache.spark.ml.param.ParamMap import org.apache.spark.ml.tuning.CrossValidator import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator import org.apache.spark.mllib.evaluation.MulticlassMetrics |
1.2 データの読み込み
まずは、データをSpark data frameとして読み込みましょう。
|
1 |
val df = sparkSession.read.option("header", "true").option("inferSchema", "true").csv("/data/titanic/train.csv") |
下記のコマンド等を使ってデータの中身を把握しましょう。
|
1 2 3 |
df.printSchema() df.show() df.groupBy("column_name").count().show() |
1.3 欠損値のハンドリング
機械学習をするにあたって、欠損値の処理をする必要があります。
今回のデータにおいてはageカラムに多くの欠損値が見られます。ageの平均を求めた後、na.fill APIを用いて欠損値に平均の値を代入します。
|
1 2 |
val meanValue = df.agg(mean(df("Age"))).first.getDouble(0) val fixedDf = df.na.fill(meanValue, Array("Age")) |
1.4 データの分割
次にrandomSplit methodを使って、データを二つに分割します. 片方はモデルのトレーニング、もう片方は精度の検証に使います。
|
1 2 3 |
val dfs = fixedDf.randomSplit(Array(0.7, 0.3)) val trainDf = dfs(0).withColumnRenamed("Survived", "label") val crossDf = dfs(1) |
1.5 カテゴリカルデータの処理
Sex(性別)やEmbarked（出港地）などのカテゴリカル（質的）データを処理していきます。
ここでは、one-hot encodingを用いてSparkMLパイプラインを作成していきます。
|
1 2 3 4 5 6 7 |
def handleCategorical(column: String): Array[PipelineStage] = { val stringIndexer = new StringIndexer().setInputCol(column) .setOutputCol(s"{column}_index") .setHandleInvalid("skip") val oneHot = new OneHotEncoder().setInputCol(s"${column}_index").setOutputCol(s"${column}_onehot") Array(stringIndexer, oneHot) } |
全てのカテゴリ変数に対してstageを作成します。
|
1 2 3 |
val genderStages = handleCategorical("Sex") val embarkedStages = handleCategorical("Embarked") val pClassStages = handleCategorical("Pclass") |
2. RandomForestを使った分類
2.1. RandomForestを使った分類器の作成
次のコードで、vector assembler stageと、分類用のrandom forest stageを作成し、pipelineの定義をします。
|
1 2 3 4 5 6 7 8 9 10 |
//columns for training val cols = Array("Sex_onehot", "Embarked_onehot", "Pclass_onehot", "SibSp", "Parch", "Age", "Fare") val vectorAssembler = new VectorAssembler().setInputCols(cols).setOutputCol("features") //algorithm stage val randomForestClassifier = new RandomForestClassifier() //pipeline val preProcessStages = genderStages ++ embarkedStages ++ pClassStages ++ Array(vectorAssembler) val pipeline = new Pipeline().setStages(preProcessStages ++ Array(randomForestClassifier)) |
2.2 modelのfitting
モデルを訓練用データに “fit”（適合）させます。
|
1 |
val model = pipeline.fit(trainDf) |
2.3. スコアの算出
モデルのパフォーマンスを計るためにスコアを算出していきます。今回の評価指標としてはaccuracy scoreを使用します。
|
1 2 3 4 5 6 7 |
def accuracyScore(df: DataFrame, label: String, predictCol: String) = { val rdd = df.select(label, predictCol).rdd.map(row ⇒ (row.getInt(0).toDouble, row.getDouble(1))) new MulticlassMetrics(rdd).accuracy } println("train accuracy with pipeline" + accuracyScore(model.transform(trainDf), "label", "prediction")) println("test accuracy with pipeline" + accuracyScore(model.transform(crossDf), "Survived", "prediction")) |
Result
|
1 2 |
train accuracy with pipeline0.8516746411483254 test accuracy with pipeline0.816793893129771 |
3. Cross-ValidationとHyper Parameter Tuning
Random forestにはいくつかの調整可能なパラメータがあります。手作業で最適なチューニングをするのは大変な作業です。今回はSparkMLのcross validation機能を使用してチューニングを行います。
3.1. Parameter gridの定義
以下のように試すべきパラメータを定義します。
|
1 2 3 4 5 |
val paramMap = new ParamGridBuilder() .addGrid(randomForestClassifier.impurity, Array("gini", "entropy")) .addGrid(randomForestClassifier.maxDepth, Array(1,2,5, 10, 15)) .addGrid(randomForestClassifier.minInstancesPerNode, Array(1, 2, 4,5,10)) .build() |
3.2 Cross-Validation
Cross-validation stageを定義して最適なパラメータを検索します。また、Cross validationを行うことで、データにover fittingしていないことも確認出来ます。
|
1 2 3 4 5 6 7 8 9 |
def crossValidation(pipeline: Pipeline, paramMap: Array[ParamMap], df: DataFrame): Model[_] = { val cv = new CrossValidator() .setEstimator(pipeline) .setEvaluator(new BinaryClassificationEvaluator) .setEstimatorParamMaps(paramMap) .setNumFolds(5) cv.fit(df) } val cvModel = crossValidation(pipeline, paramMap, trainDf) |
2.3の時のスコアに比べてスコアが改善されたのが分かります。
|
1 2 |
train accuracy with cross validation0.8787878787878788 test accuracy with cross validation 0.8577099236641222 |
4. Submit Fileの生成
Kaggle competitionのように提出用のテスト結果ファイルを生成してみましょう。
|
1 2 3 |
val testDf = sparkSession.read.option("header", "true").option("inferSchema", "true").csv("src/main/resources/titanic/test.csv") val fareMeanValue = df.agg(mean(df("Fare"))).first.getDouble(0) val fixedOutputDf = testDf.na.fill(meanValue, Array("age")).na.fill(fareMeanValue, Array("Fare")) |
|
1 2 3 4 5 6 |
def generateOutputFile(testDF: DataFrame, model: Model[_]) = { val scoredDf = model.transform(testDF) val outputDf = scoredDf.select("PassengerId", "prediction") val castedDf = outputDf.select(outputDf("PassengerId"), outputDf("prediction").cast(IntegerType).as("Survived")) castedDf.write.format("csv").option("header", "true").mode(SaveMode.Overwrite).save("src/main/resources/output/") } |
参考
あなたも、Avintonでこのような最先端技術に日常的に
取り組んでみませんか？
優秀なエンジニアの成長を導きながら、AIやビッグデータなどの最先端技術を活用していくことが私たちのビジョンです。
Avintonの充実した技術研修でスキルアップを図り、あなたのキャリア目標を達成しませんか？
