Machine Learning Uygulaması: Modelin Oluşturulması

Bir önceki yazıda Machine Learning probleminden ve kullanacağımız araçlardan bahsettik. Şimdi ise bu araçları ve veri setimizi kullanarak modelimizi oluşturacağız.

Hazırlık

Apache Spark için cluster kurmayacağımızdan ve sadece bellekte ayağa kaldıracağımızdan yapmamız gereken çok fazla bir şey olmayacak. Sadece Java ile basit bir konsol uygulaması yazacağız. Bu konsol uygulaması aşağıdaki adımları takip edecek:

Bellekte Spark uygulamasını ayağa kaldır
Veri setini Spark’a yükle
Veri setini normalizasyon fonksiyonunu kullanarak veriyi istenilen duruma getir
Machine Learning algoritmasını kullanarak modeli oluştur
Modeli test seti ile test et ve tutarlılığına bak
Eğer tutarlılığı istenilen düzeydeyse modeli kaydet

Burada detayına inmediğimiz iki konu var. Birincisi hangi Machine Learning algoritmasının kullanılacağı. İkincisi ise normalizasyon fonksiyonumuzun ne olacağı. Kodumuzu geliştirmeye başlamadan önce bu iki konuya açıklık getirelim.

ML Algoritmasının Seçimi

Çok fazla Machine Learning Algoritmalarının detayına girmeyeceğim. Ama en temelde Machine Learning algoritmaları iki temel gruba ayrılıyor Supervised Learning ve Unsupervised Learning. Unsupervised Learning’te elinizdeki veri seti etiketlenmemiş oluyor. Algoritmalar bu veri setini sadece gruplara ayırıyor. Sonrasında siz hangi grubun ne olduğuna karar veriyorsunuz. Supervised Learning‘te ise elinizde bir veri seti var ve veri setinde tüm veriler etiketlenmiş. Siz bu etiketler üzerinden modelinizi oluşturuyorsunuz. Bizim kullanacağımız veri setine bakacak olursanız her bir kalemin hangi sınıfa ait olduğu bilgisi var. Bizde bu sebepten Supervised algoritmalara bakacağız.

Çözmeye çalıştığımız problemimizle ilgili diğer bildiğimiz şey ise sınıflandırmaya çalıştığımız. Bu sebepten Classification algoritmalarına göz atmalıyız. Problemimiz basit bir problem olduğu için, hemen hemen tüm Classification algoritmaları doğru sonuç üretecektir. Ben Apache Spark da desteklediği için Logistic Regression algoritmasını kullanacağım.

Machine Learning algoritmaları hakkında daha detaylı bilgiyi buradan alabilirsiniz. Logistic Regression ile ilgili daha detaylı bilgiye ise buradan ulaşabilirsiniz.

Değerleri Normalize Etmek

Önceki yazımda 6 farklı etkeni kullanacağımızdan ve bu etkenlerin “ortalamanın üzerinde” (Positive), “ortalamanın altında” (Negative) ve “ortalama” (Average) şeklinde değerler aldıklarından bahsetmiştim. Her ne kadar bazı Regression algoritmaları, normalizasyonu zorunlu kılsa da aslında Logistic Regression için normalizasyon zorunlu değil. Fakat bizim verimiz matematiksel işlemlere tabi tutulacak cinsten olmadığı için normalizasyon yapmaya mecbur kalıyoruz.

Verimizin ham hali aşağıdaki gibi

P,N,P,A,A,P,NB
N,P,A,P,A,P,NB
N,P,A,A,P,N,NB
A,N,N,N,N,A,B
P,N,N,N,N,N,B

P,N,P,A,A,P,NB

N,P,A,P,A,P,NB

N,P,A,A,P,N,NB

A,N,N,N,N,A,B

P,N,N,N,N,N,B

P (Positive), N (Negative) ve A (Average) değerlerini normalize etmek için aşağıdaki gibi bir method kullanmayı düşünüyorum.

private static double normalizeFeature(String data) {
    if ("P".equals(data)) return 1.0;
    if ("A".equals(data)) return 0.0;
    if ("N".equals(data)) return -1.0;

    throw new IllegalArgumentException("Unexpected data: " + data);
}

private static double normalizeFeature(String data) {

if ("P".equals(data)) return 1.0;

if ("A".equals(data)) return 0.0;

if ("N".equals(data)) return -1.0;

throw new IllegalArgumentException("Unexpected data: " + data);

}

Neden 1, 0 ve -1 seçtiğimi soracak olursanız, çok teknik ya da detaylı bir cevap vermem güç. Mesela bazı Regression uygulamaları verilerinizin -1 ve 1 (Feature Scaling) aralığında daha iyi sonuç verir. Fakat Logistic Regression için bu geçerli değil. Ama yine de hem aynı notasyonu takip etmek hem de değerlere (Positive, Negative) tam denk geldiğinden 1, 0 ve -1 değerlerini kullandım. Farklı bir değerlerle de benzer sonuçlara alabilirsiniz.

Ek olarak burada tekrar hatırlatmak istiyorum. Bir sonraki yazıda servisimizi yazarken de aynı normalizasyon fonksiyonunu kullanmalıyız.

Son olarak sınıflarımızı inceleyelim. Bildiğiniz üzere iki temel sınıfımız var: İflas eden (Bankruptcy) / iflas etmeyen (Non-Bankruptcy). Bunlar için de sayı atamalıyız. Hangisine neyi atadığınız çok önemli değil. Notasyon genelde sıfırdan başlar ve ardışık devam eder. Bende benzer notasyon kullanıp Bankruptcy için 0 Non-Bankruptcy için 1 atayacağım. Daha sonra uygulamamızı yazıp yeni veriler için modelimizi çalıştırdığımızda, modelimiz bize 0 ya da 1 dönecek. Bu değerleri Bankruptcy ya da Non-Bankruptcy olarak yorumlamak yine bize düşüyor.

Model Üretmek

Öncelikle pom.xml‘inize bağımlılıklarımızı ekleyelim.

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    ...

    <dependencies>
        ...
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>2.0.0</version>
        </dependency>
        ...
    </dependencies>

    ...
</project>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

...

...

<groupId>org.apache.spark</groupId>

<artifactId>spark-mllib_2.11</artifactId>

</dependency>

...

</dependencies>

...

</project>

Ne yazık ki Spark’ın bağımlılıkları biraz fazla. Bunun temel nedeni, Spark’ın Scala kullanılarak geliştirilmiş olması.

Sonrasında önceki yazıda indirdiğiniz arşivden çıkan Qualitative_Bankruptcy.data.txt isimli dosyayı src/main/resources dizinine kopyalıyorsunuz. Ardından QualitativeBankruptcyModelGenerator isimli java sınıfını yaratıp içeriğini aşağıdaki gibi yapıyoruz.

QualitativeBankruptcyModelGenerator.java

package com.bahadirakin.ml;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.classification.LogisticRegressionModel;
import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS;
import org.apache.spark.mllib.evaluation.MulticlassMetrics;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.regression.LabeledPoint;
import scala.Tuple2;

import java.util.Collections;

public class QualitativeBankruptcyModelGenerator {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("QualitativeBankruptcyModelGenerator")
                .setMaster("local");
        final SparkContext sparkContext = new SparkContext(conf);
        final JavaSparkContext sc = new JavaSparkContext(sparkContext);

        final String path = "src/main/resources/Qualitative_Bankruptcy.data.txt";
        final JavaRDD<String> textFile = sc.textFile(path, 1);
        System.out.println("Data Count: " + textFile.count());

        final JavaRDD<LabeledPoint> data = textFile.map(line -> {
            final String[] split = line.split(",");
            final double label = normalizeLabel(split[split.length - 1]);
            final double[] doubles = new double[split.length - 1];
            for (int i = 0; i < split.length - 1; i++) {
                doubles[i] = normalizeFeature(split[i]);
            }
            final Vector features = Vectors.dense(doubles);
            return new LabeledPoint(label, features);
        });

        data.take(10).forEach(System.out::println);

        // Split initial RDD into two... [60% training data, 40% testing data].
        JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.6, 0.4}, 11L);
        JavaRDD<LabeledPoint> training = splits[0].cache();
        JavaRDD<LabeledPoint> test = splits[1];

        // Run training algorithm to build the model.
        final LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
                .setNumClasses(2)
                .run(training.rdd());

        // Compute raw scores on the test set.
        JavaRDD<Tuple2<Object, Object>> predictionAndLabels = test.map(p -> {
            Double prediction = model.predict(p.features());
            return new Tuple2<>(prediction, p.label());
        });

        // Get evaluation metrics.
        final MulticlassMetrics metrics = new MulticlassMetrics(predictionAndLabels.rdd());
        System.out.println("Accuracy = " + metrics.accuracy());

        // Save model
        final String modelLink = "target/model/Qualitative_Bankruptcy_Model";
        model.save(sparkContext, modelLink);


        sparkContext.stop();
        sc.stop();
    }

    private static double normalizeFeature(String data) {
        if ("P".equals(data)) return 1.0;
        if ("A".equals(data)) return 0.0;
        if ("N".equals(data)) return -1.0;

        throw new IllegalArgumentException("Unexpected data: " + data);
    }

    private static double normalizeLabel(String data) {
        if ("NB".equals(data)) return 1.0;
        if ("B".equals(data)) return 0.0;

        throw new IllegalArgumentException("Unexpected data: " + data);
    }
}

package com.bahadirakin.ml;

import org.apache.spark.SparkConf;

import org.apache.spark.SparkContext;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.mllib.classification.LogisticRegressionModel;

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS;

import org.apache.spark.mllib.evaluation.MulticlassMetrics;

import org.apache.spark.mllib.linalg.Vector;

import org.apache.spark.mllib.linalg.Vectors;

import org.apache.spark.mllib.regression.LabeledPoint;

import scala.Tuple2;

import java.util.Collections;

public class QualitativeBankruptcyModelGenerator {

public static void main(String[] args) {

SparkConf conf = new SparkConf()

.setAppName("QualitativeBankruptcyModelGenerator")

.setMaster("local");

final SparkContext sparkContext = new SparkContext(conf);

final JavaSparkContext sc = new JavaSparkContext(sparkContext);

final String path = "src/main/resources/Qualitative_Bankruptcy.data.txt";

final JavaRDD<String> textFile = sc.textFile(path, 1);

System.out.println("Data Count: " + textFile.count());

final JavaRDD<LabeledPoint> data = textFile.map(line -> {

final String[] split = line.split(",");

final double label = normalizeLabel(split[split.length - 1]);

final double[] doubles = new double[split.length - 1];

for (int i = 0; i < split.length - 1; i++) {

doubles[i] = normalizeFeature(split[i]);

}

final Vector features = Vectors.dense(doubles);

return new LabeledPoint(label, features);

});

data.take(10).forEach(System.out::println);

// Split initial RDD into two... [60% training data, 40% testing data].

JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.6, 0.4}, 11L);

JavaRDD<LabeledPoint> training = splits[0].cache();

JavaRDD<LabeledPoint> test = splits[1];

// Run training algorithm to build the model.

final LogisticRegressionModel model = new LogisticRegressionWithLBFGS()

.setNumClasses(2)

.run(training.rdd());

// Compute raw scores on the test set.

JavaRDD<Tuple2<Object, Object>> predictionAndLabels = test.map(p -> {

Double prediction = model.predict(p.features());

return new Tuple2<>(prediction, p.label());

});

// Get evaluation metrics.

final MulticlassMetrics metrics = new MulticlassMetrics(predictionAndLabels.rdd());

System.out.println("Accuracy = " + metrics.accuracy());

// Save model

final String modelLink = "target/model/Qualitative_Bankruptcy_Model";

model.save(sparkContext, modelLink);

sparkContext.stop();

sc.stop();

}

private static double normalizeFeature(String data) {

if ("P".equals(data)) return 1.0;

if ("A".equals(data)) return 0.0;

if ("N".equals(data)) return -1.0;

throw new IllegalArgumentException("Unexpected data: " + data);

}

private static double normalizeLabel(String data) {

if ("NB".equals(data)) return 1.0;

if ("B".equals(data)) return 0.0;

throw new IllegalArgumentException("Unexpected data: " + data);

}

Yukarıdaki sınıfı çalıştırdığımızda modelimizin target/model/Qualitative_Bankruptcy_Model altında yaratılmış olduğunu göreceksiniz. Bir sonraki yazıda bu dizine ihtiyacımız olacak.

Kısaca yukarıda yapılan işlemlerden de bahsetmek istiyorum.

SparkConf ile Spark ayarlarımızı yapıyoruz. Burada dikkat etmenizi istediğim nokta setMaster kısmı. Eğer Cluster’ımız olsaydı ve biz sorgumuzu cluster’a göndermek isteseydik burada Master’ın adresini verecektik. Yok eğer Cluster’ımıza Jar olarak yüklemek isteseydik, setMaster komutuna gerek kalmayacaktı.
SparkContext ve JavaSparkContext: Spark Context dilden bağımsız Spark ile ilgil işlemleri yapmanıza olanak sağlıyor. Ama Java tabanlı işlemler yapacaksanız JavaSparkContext’e ihtiyacınız var.
RDD (Resilient Distributed Dataset), Spark’ın temel nesnesi gibi düşünebilirsiniz. Spark bu nesneler üzerinden veri setinizi cluster’ı üzerinde dağıtıyor.
Verimizin %40’nı test seti olarak kullanıyoruz. Hangi kısmın test seti olarak kullanılacağı rastgele seçiliyor.
Hangi algoritmayı kullandığımızla veri tamamen ayrıştırılmış durumda. Bu sebepten farklı algoritmaları da rahatlıkla deneyebilirsiniz.
Biz Logistic Regression yapacağımız için LogisticRegressionModel modelini kullanıyoruz. Dikkat ettiyseniz modelimizi oluşturacak algoritmayı seçerken, sonuç kümemizde kaç farklı sınıf olacağını da belirtiyoruz.
Kodu çalıştırdığınız tutarlılığı (Accuracy) da ekrana basıyor. Tutarlılık bende 0.9574468085106383 çıktı. Kötü bir tutarlılık değil. Fakat çok iyi olduğu da söylenemez ama kabul edilebilir.

Son

Artık modelimizi nasıl oluşturacağımızı da biliyoruz. Bir sonraki yazıda modelimizi servislerden aldığımız yeni verilerle nasıl kullanacağımıza bakacağız.