Mtl training

src/main/java/ml/shifu/shifu/container/obj/ModelTrainConf.java

@@ -41,7 +41,7 @@ public class ModelTrainConf {
      * @author Zhang David (pengzhang@paypal.com)
      */
     public static enum ALGORITHM {
-        NN, LR, SVM, DT, RF, GBT, TENSORFLOW, WDL
+        NN, LR, SVM, DT, RF, GBT, TENSORFLOW, WDL, MTL
     }
 
     /**

src/main/java/ml/shifu/shifu/core/MTLModel.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright [2013-2019] PayPal Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ml.shifu.shifu.core;
+
+import ml.shifu.shifu.core.dtrain.mtl.IndependentMTLModel;
+import org.encog.ml.BasicML;
+import org.encog.ml.MLRegression;
+import org.encog.ml.data.MLData;
+import org.encog.ml.data.basic.BasicMLData;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * @author haillu
+ */
+public class MTLModel extends BasicML implements MLRegression {
+    private IndependentMTLModel independentMTLModel;
+
+    public MTLModel(IndependentMTLModel independentMTLModel) {
+        this.independentMTLModel = independentMTLModel;
+    }
+
+    @Override
+    public void updateProperties() {
+        // No need to implement
+    }
+
+    @Override
+    public MLData compute(MLData input) {
+        double[] result = independentMTLModel.compute(input.getData());
+        return new BasicMLData(result);
+    }
+
+    @Override
+    public int getInputCount() {
+        return independentMTLModel.getMtl().getInputSize();
+    }
+
+    @Override
+    public int getOutputCount() {
+        return independentMTLModel.getMtl().getTaskNumber();
+    }
+
+    public static MTLModel loadFromStream(InputStream input) throws IOException {
+        return new MTLModel(IndependentMTLModel.loadFromStream(input));
+    }
+
+    public static MTLModel loadFromStream(InputStream input, boolean isRemoveNameSpace) throws IOException {
+        return new MTLModel(IndependentMTLModel.loadFromStream(input, isRemoveNameSpace));
+    }
+
+}

src/main/java/ml/shifu/shifu/core/Scorer.java

@@ -428,6 +428,28 @@ public MLData call() {
                         log.error("error in model evaluation", e);
                     }
                 }
+            } else if (model instanceof MTLModel){
+                final MTLModel mtl = (MTLModel) model;
+                if(mtl.getInputCount() != pair.getInput().size()) {
+                    throw new RuntimeException("MTL and input size mismatch: mtl input Size = " + mtl.getInputCount()
+                            + "; data input Size = " + pair.getInput().size());
+                }
+
+                Callable<MLData> callable = new Callable<MLData>() {
+                    @Override
+                    public MLData call() {
+                        return new BasicMLData(mtl.compute(pair.getInput()));
+                    }
+                };
+                if(multiThread) {
+                    tasks.add(callable);
+                } else {
+                    try {
+                        modelResults.add(callable.call());
+                    } catch (Exception e) {
+                        log.error("error in MTL model evaluation", e);
+                    }
+                }
             } else {
                 throw new RuntimeException("unsupport models");
             }
@@ -525,6 +547,8 @@ public int compare(String o1, String o2) {
                     scores.add(toScore(score.getData(0)));
                 } else if(model instanceof WDLModel) {
                     scores.add(toScore(score.getData(0)));
+                } else if (model instanceof MTLModel){
+                    scores.add(toScore(score.getData(0)));
                 } else {
                     throw new RuntimeException("unsupport models");
                 }

src/main/java/ml/shifu/shifu/core/dtrain/CommonConstants.java

@@ -146,6 +146,8 @@ public interface CommonConstants {
 
     public static final int WDL_FORMAT_VERSION = 1;
 
+    public static final int MTL_FORMAT_VERSION = 1;
+
     public static final int DEFAULT_EMBEDING_OUTPUT = 8;
 
     public static final String WIDE_ENABLE = "wideEnable";

src/main/java/ml/shifu/shifu/core/dtrain/SerializationType.java

@@ -0,0 +1,27 @@
+package ml.shifu.shifu.core.dtrain;
+
+import java.util.Arrays;
+
+/**
+ * @author haillu
+ */
+public enum SerializationType {
+    /**
+     * Serialize types, each of them including different serialize scope
+     */
+    WEIGHTS(0), GRADIENTS(1), MODEL_SPEC(2), ERROR(-1);
+
+    int value;
+
+    SerializationType(int type) {
+        this.value = type;
+    }
+
+    public static SerializationType getSerializationType(int value) {
+        return Arrays.stream(values()).filter(type -> type.value == value).findFirst().orElse(SerializationType.ERROR);
+    }
+
+    public int getValue() {
+        return this.value;
+    }
+}

src/main/java/ml/shifu/shifu/core/dtrain/mtl/BinaryMTLSerializer.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright [2013-2019] PayPal Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ml.shifu.shifu.core.dtrain.mtl;
+
+import ml.shifu.shifu.container.obj.ColumnConfig;
+import ml.shifu.shifu.container.obj.ModelConfig;
+import ml.shifu.shifu.core.Normalizer;
+import ml.shifu.shifu.core.dtrain.CommonConstants;
+import ml.shifu.shifu.core.dtrain.DTrainUtils;
+import ml.shifu.shifu.core.dtrain.SerializationType;
+import ml.shifu.shifu.core.dtrain.StringUtils;
+import ml.shifu.shifu.core.dtrain.nn.NNColumnStats;
+import ml.shifu.shifu.util.CommonUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.zip.GZIPOutputStream;
+
+/**
+ * @author haillu
+ */
+public class BinaryMTLSerializer {
+    public static void save(ModelConfig modelConfig, List<List<ColumnConfig>> mtlColumnConfigLists,
+            MultiTaskLearning mtl, FileSystem fs, Path output) throws IOException {
+        DataOutputStream fos = null;
+        try {
+            fos = new DataOutputStream(new GZIPOutputStream(fs.create(output)));
+
+            // version
+            fos.writeInt(CommonConstants.MTL_FORMAT_VERSION);
+            // Reserved two double field, one double field and one string field
+            fos.writeDouble(0.0f);
+            fos.writeDouble(0.0f);
+            fos.writeDouble(0.0d);
+            fos.writeUTF("Reserved field");
+
+            // write normStr
+            String normStr = modelConfig.getNormalize().getNormType().toString();
+            StringUtils.writeString(fos, normStr);
+
+            // write task number.
+            fos.writeInt(mtlColumnConfigLists.size());
+
+            for(List<ColumnConfig> ccs: mtlColumnConfigLists) {
+                // compute columns needed
+                Map<Integer, String> columnIndexNameMapping = getIndexNameMapping(ccs);
+
+                // write column stats to output
+                List<NNColumnStats> csList = new ArrayList<>();
+                for(ColumnConfig cc: ccs) {
+                    if(columnIndexNameMapping.containsKey(cc.getColumnNum())) {
+                        NNColumnStats cs = new NNColumnStats();
+                        cs.setCutoff(modelConfig.getNormalizeStdDevCutOff());
+                        cs.setColumnType(cc.getColumnType());
+                        cs.setMean(cc.getMean());
+                        cs.setStddev(cc.getStdDev());
+                        cs.setColumnNum(cc.getColumnNum());
+                        cs.setColumnName(cc.getColumnName());
+                        cs.setBinCategories(cc.getBinCategory());
+                        cs.setBinBoundaries(cc.getBinBoundary());
+                        cs.setBinPosRates(cc.getBinPosRate());
+                        cs.setBinCountWoes(cc.getBinCountWoe());
+                        cs.setBinWeightWoes(cc.getBinWeightedWoe());
+
+                        // TODO cache such computation
+                        double[] meanAndStdDev = Normalizer.calculateWoeMeanAndStdDev(cc, false);
+                        cs.setWoeMean(meanAndStdDev[0]);
+                        cs.setWoeStddev(meanAndStdDev[1]);
+                        double[] weightMeanAndStdDev = Normalizer.calculateWoeMeanAndStdDev(cc, true);
+                        cs.setWoeWgtMean(weightMeanAndStdDev[0]);
+                        cs.setWoeWgtStddev(weightMeanAndStdDev[1]);
+
+                        csList.add(cs);
+                    }
+                }
+
+                fos.writeInt(csList.size());
+                for(NNColumnStats cs: csList) {
+                    cs.write(fos);
+                }
+
+                Map<Integer, Integer> columnMapping = DTrainUtils.getColumnMapping(ccs);
+                fos.writeInt(columnMapping.size());
+                for(Map.Entry<Integer, Integer> entry: columnMapping.entrySet()) {
+                    fos.writeInt(entry.getKey());
+                    fos.writeInt(entry.getValue());
+                }
+
+            }
+
+            // persist multi task learning Model
+            mtl.write(fos, SerializationType.MODEL_SPEC);
+        } finally {
+            IOUtils.closeStream(fos);
+        }
+    }
+
+    private static Map<Integer, String> getIndexNameMapping(List<ColumnConfig> columnConfigList) {
+        Map<Integer, String> columnIndexNameMapping = new HashMap<>(columnConfigList.size());
+        for(ColumnConfig columnConfig: columnConfigList) {
+            if(columnConfig.isFinalSelect()) {
+                columnIndexNameMapping.put(columnConfig.getColumnNum(), columnConfig.getColumnName());
+            }
+        }
+
+        if(columnIndexNameMapping.size() == 0) {
+            boolean hasCandidates = CommonUtils.hasCandidateColumns(columnConfigList);
+            for(ColumnConfig columnConfig: columnConfigList) {
+                if(CommonUtils.isGoodCandidate(columnConfig, hasCandidates)) {
+                    columnIndexNameMapping.put(columnConfig.getColumnNum(), columnConfig.getColumnName());
+                }
+            }
+        }
+        return columnIndexNameMapping;
+    }
+}