[Apache Spark] Machine Learning from Disaster: Naive Bayes

Pre-processing the data

First we need to  read the data from csv and implement the data preparation step from the last post.

Reading unstructured data in Spark is very easy, you can read in every file with the textFile($path) function from the SparkContext. As we got structured csv data, we want to read it with the new DataFrame/Dataset API and not as a simple Textfile. To do so we use a 3rd party library from databricks (spark-csv).

    val df = sqlContext.read
      .format("com.databricks.spark.csv")
      .option("header", "true") // Use first line of all files as header
      .option("inferSchema", "true") // Automatically infer data types
      .load("src/main/resources/titanic_data/train.csv")

Getting some statistics

We know that Fare and Age columns have null values and we want to fill in the mean value in that case. Spark ML has a Statistics class, which offers this calculations for RDDs of type Vector. So we create dense vectors from Fare and Age for each row.

    
    val statsDf = df.map { row =>
      Vectors.dense(row.getAs("Fare"), row.getAs("Age"))
    }
    val summary: MultivariateStatisticalSummary = Statistics.colStats(statsDf)
    val meanFare =  summary.mean(0)
    val meanAge =  summary.mean(1)

User defined functions

In Spark/Scala things are immutable by default, we always create new Dataframes or RDDs after manipulating the data. In the RDD context we can simply use map to manipulate Data. When working with Dataframes (structured data) we need to use user defined functions. A user defined function can be defined for multiple input columns or just for one. We want to only filter null values, so we only use one.

After defining the UDFs we select the columns (SQL like) and use the udf as a function in the select statement. Alias does name the new column (which replaces the old one in this case).

    
      val normFare = udf((d: String) => d match {
      case null => Some(meanFare)
      case s => Some(s.toDouble)
    })

    val normAge = udf((d: String) => d match {
      case null => Some(meanAge)
      case s => Some(s.toDouble)
    })

    // select the columns we need and apply the udfs
    val preprocessed = df.select(df("Survived"), normFare(df("Fare")).alias("Fare"), normAge(df("Age")).alias("Age"), df("Pclass"))

Training

To train the bayes model we need LabeledPoints, which represent the annotated data. Each labeled point is a entry in the DataFrame and consists of the label and the feature vector. The label is the class, un out case there are only 2 classes survived and not-survived. If you have more labels use whole numbers to represent them.

The NaiveBayes supports some optional parameters. You can specify the lambda and the method, which is used. Multinomial is what we want to use, the Bernoulli model would be only accept 0 and 1 as vector values (occurrences).

    
      val mappedDf = df.map(row =>
      (row.getAs[Int]("Survived"), row.getAs[Double]("Fare"), row.getAs[Int]("Pclass"), row.getAs[Double]("Age")))

    val labledData = mappedDf.map { case (survived, fare, pclass, age) =>
      LabeledPoint(survived, Vectors.dense(fare, pclass, age))
    }
    naiveBayesModel = NaiveBayes.train(labledData, lambda = 1.0, modelType = "multinomial")

Predictions

Now we have trained the model and can predict survival of passengers from the validation/test data set. We read the data from the test set like the training set and also create the feature vectors. Then we use the predict method, which returns the class. In this case we don´t use udfs just to show how you can convert between RDD and Dataframes.

    
       val resultRDD = df.map { row =>
      val denseVecor = Vectors.dense(row.getAs[Double]("Fare"), row.getAs[Int]("Pclass"), row.getAs[Double]("Age"))
      val result = naiveBayesModel.predict(denseVecor)
      Row.fromTuple((row.getAs[Int]("PassengerId"), result.toInt))
    }
val customSchema = StructType(Array(
      StructField("PassengerId", IntegerType, false),
      StructField("Survived", DoubleType, false))
    )
    val resultDf = sqlContext.createDataFrame(resultRDD, customSchema)

Save results

Now we have the result and want to save it to a csv file for Kaggle submission. To avoid, that we write the csv by our self we converted the RDD to structured data (DataFrame) and can use the databricks csv library.

    
resultDf
      //  only one file (one partition)
      .coalesce(1)
      .write
      .format("com.databricks.spark.csv")
      .option("header", "true")
      .save("results/out_" + System.currentTimeMillis())

After the submission, we can see the result isn´t that bad, but its also not very good. So what about the gender? Take a look at the github code!

Thanks for reading! Please comment, if you have any thoughts!