myui

import sys

from pyspark.context import SparkContext
from numpy import array, random as np_random
from sklearn import linear_model as lm
from sklearn.base import copy
from scipy import sparse as sp

#MAX_FEATURES=1000
MAX_FEATURES=16777216

myui

import sys

from pyspark.context import SparkContext
from numpy import array, random as np_random
from sklearn import linear_model as lm
from sklearn.base import copy

ITERATIONS = 5
np_random.seed(seed=42)

myui

#! /usr/bin/env python

import sys

from sklearn.externals import joblib
from scipy import sparse as sp

MAX_FEATURES=16777216

def predict(sgd, line):

myui

create or replace view training2 as
select
  rowid,
  clicks,
  (impression - clicks) as noclick,
  mhash(concat("1_", displayurl)) as displayurl, 
  mhash(concat("2_", adid)) as adid, 
  ...
  -1 as bias
from (

myui

T add(T item) {
    T old = null;
    if(position < numSamples) {// reservoir not yet full, just append
        samples[position] = item;
    } else {// find a item to replace
        int replaceIndex = rand.nextInt(position + 1);
        if(replaceIndex < numSamples) {// replacement opportunity decreases over a time
            old = samples[replaceIndex];
            samples[replaceIndex] = item;
        }

myui

public final class TrainNewsGroups {

  public static void main(String[] args) throws IOException {
    File base = new File(args[0]);

    Multiset<String> overallCounts = HashMultiset.create();

    int leakType = 0;
    if (args.length > 1) {
      leakType = Integer.parseInt(args[1]);

myui

import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.mllib.classification.LogisticRegressionWithSGD

val training = MLUtils.loadLibSVMFile(sc, "hdfs://dm01:8020/user/hive/warehouse/kdd12track2.db/training_libsvmfmt_10k",  multiclass = false, numFeatures = 16777216, minPartitions = 64)
//val training = MLUtils.loadLibSVMFile(sc, "hdfs://dm01:8020/user/hive/warehouse/kdd12track2.db/training_libsvmfmt_10k",  multiclass = false)

val model = LogisticRegressionWithSGD.train(training, numIterations = 1)
//val model = LogisticRegressionWithSGD.train(training, numIterations = 20)

myui

The dataset used in the Evaluation

http://www.csie.ntu.edu.tw/~cjlin/libsvmtools/datasets/binary.html#news20.binary

head -1000 news20.binary | sed 's/+1/1/g' | sed 's/-1/0/g' > news20.binary.1000
sort -R news20.binary > news20.random
head -1000 news20.random | sed 's/+1/1/g' | sed 's/-1/0/g' > news20.random.1000

Evaluated code

myui

liblinear on spark

http://www.csie.ntu.edu.tw/~cjlin/libsvmtools/distributed-liblinear/spark/running_spark_liblinear.html

SPARK_CLASSPATH=/opt/spark-1.0.0/extlib/spark-liblinear-1.94.jar ./bin/spark-shell

sc.addJar("./extlib/spark-liblinear-1.94.jar")
import tw.edu.ntu.csie.liblinear._

myui

Increasing LBFGS passes

[GD: 1 iters LBFGS: 20 iters mapper: 215]

> real    8m13.805s
> AUC  : 0.707109
> NWMAE: 0.049646
> WRMSE: 0.158077