[GitHub] [submarine] pingsutw commented on a change in pull request #300: SUBMARINE-485. [SDK] Add PyTorch implementation of DeepFM model

[GitBox <gi...@apache.org>]

pingsutw commented on a change in pull request #300:
URL: https://github.com/apache/submarine/pull/300#discussion_r430400562



##########
File path: submarine-sdk/pysubmarine/submarine/ml/tensorflow/model/base_tf_model.py
##########
@@ -67,8 +67,8 @@ def train(self, train_input_fn=None, eval_input_fn=None, **kwargs):
 
     def evaluate(self, eval_input_fn=None, **kwargs):
         """
-        Evaluates a pre-defined Tensorflow estimator model with given evaluate data
-        :param eval_input_fn: A function that provides input data for evaluating.
+        Evaluates a pre-defined Tensorflow estimator model with given evaluate input
+        :param eval_input_fn: A function that provides input input for evaluating.

Review comment:
       ```suggestion
           Evaluates a pre-defined Tensorflow estimator model with given evaluate data
           :param eval_input_fn: A function that provides input data for evaluating.
   ```

##########
File path: submarine-sdk/pysubmarine/example/pytorch/run_ctr.py
##########
@@ -0,0 +1,41 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+
+from submarine.ml.pytorch.model.ctr import DeepFM
+
+import argparse
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-conf", help="a JSON configuration file for FM", type=str)

Review comment:
       ```suggestion
           "-conf", help="a JSON configuration file for DeepFM", type=str)
   ```

##########
File path: submarine-sdk/pysubmarine/submarine/ml/pytorch/model/base_pytorch_model.py
##########
@@ -0,0 +1,163 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+from abc import ABC
+import logging
+
+from submarine.ml.pytorch.metric import get_metric_fn
+from submarine.ml.pytorch.registries import input_fn_registry
+from submarine.utils.fileio import write_file
+from submarine.ml.abstract_model import AbstractModel
+import torch
+from torch import distributed
+from torch.nn.parallel import DistributedDataParallel
+
+import os
+import io
+
+from submarine.ml.pytorch.optimizer import get_optimizer
+from submarine.ml.pytorch.loss import get_loss_fn
+from submarine.ml.pytorch.parameters import default_parameters
+from submarine.utils.env import get_from_registry, get_from_json, get_from_dicts
+from submarine.utils.pytorch_utils import get_device
+
+logger = logging.getLogger(__name__)
+
+
+# pylint: disable=W0221
+class BasePyTorchModel(AbstractModel, ABC):
+    def __init__(self, params=None, json_path=None):
+        super().__init__()
+        self.params = get_from_dicts(params, default_parameters)
+        self.params = get_from_json(json_path, self.params)
+        self._sanity_check()
+        logging.info("Model parameters : %s", self.params)
+        self.input_type = self.params['input']['type']
+
+        self.init_process_group()
+        self.model = DistributedDataParallel(
+            self.model_fn(self.params).to(get_device(self.params)))
+        self.optimizer = get_optimizer(key=self.params['optimizer']['name'])(
+            params=self.model.parameters(),
+            **self.params['optimizer']['kwargs']
+        )
+        self.loss = get_loss_fn(key=self.params['loss']['name'])(
+            **self.params['loss']['kwargs'])
+        self.metric = get_metric_fn(key=self.params['output']['metric'])
+
+    def init_process_group(self):
+        distributed.init_process_group(
+            backend=os.environ.get('backend', distributed.Backend.GLOO),
+            init_method=os.environ.get('INIT_METHOD', 'tcp://127.0.0.1:23456'),
+            world_size=int(os.environ.get('WORLD', 1)),
+            rank=int(os.environ.get('RANK', 0))
+        )
+
+    def __del__(self):
+        distributed.destroy_process_group()
+
+    def train(self, train_loader):
+        for _, batch in enumerate(train_loader):
+            sample, target = batch
+            output = self.model(sample)
+            loss = self.loss(output, target)
+            loss.backward()
+            self.optimizer.zero_grad()
+            self.optimizer.step()
+
+    def evaluate(self):
+        outputs = []
+        targets = []
+
+        valid_loader = get_from_registry(
+            self.input_type, input_fn_registry)(
+            filepath=self.params['input']['valid_data'],
+            **self.params['training'])()
+
+        with torch.no_grad():
+            for _, batch in enumerate(valid_loader):
+                sample, target = batch
+                output = self.model(sample)
+
+                outputs.append(output)
+                targets.append(target)
+
+        return self.metric(
+            torch.cat(targets, dim=0).cpu().numpy(),
+            torch.cat(outputs, dim=0).cpu().numpy()
+        )
+
+    def predict(self):
+        outputs = []
+
+        test_loader = get_from_registry(
+            self.input_type, input_fn_registry)(
+            filepath=self.params['input']['test_data'],
+            **self.params['training'])()
+
+        with torch.no_grad():
+            for _, batch in enumerate(test_loader):
+                sample, _ = batch
+                output = self.model(sample)
+                outputs.append(output)
+        # TODO: fix this
+        return torch.cat(outputs, dim=0).cpu().numpy()
+
+    def fit(self):
+        # TODO: fix this
+        best_eval_score = 0.0
+        train_loader = get_from_registry(
+            self.input_type, input_fn_registry)(
+            filepath=self.params['input']['train_data'],
+            **self.params['training'])()
+
+        for epoch in range(self.params['training']['num_epochs']):
+            train_loader.sampler.set_epoch(epoch)
+            self.train(train_loader)
+            eval_score = self.evaluate()
+            # TODO: fix this

Review comment:
       Need to describe more detail here.
   `# TODO(andrewhsiehth): Fix .....`

##########
File path: submarine-sdk/pysubmarine/submarine/ml/tensorflow/model/base_tf_model.py
##########
@@ -43,9 +43,9 @@ def __init__(self, model_params=None, json_path=None):
 
     def train(self, train_input_fn=None, eval_input_fn=None, **kwargs):
         """
-        Trains a pre-defined tensorflow estimator model with given training data
-        :param train_input_fn: A function that provides input data for training.
-        :param eval_input_fn: A function that provides input data for evaluating.
+        Trains a pre-defined tensorflow estimator model with given training input
+        :param train_input_fn: A function that provides input input for training.
+        :param eval_input_fn: A function that provides input input for evaluating.

Review comment:
       ```suggestion
           Trains a pre-defined tensorflow estimator model with given training data
           :param train_input_fn: A function that provides input data for training.
           :param eval_input_fn: A function that provides input data for evaluating.
   ```

##########
File path: submarine-sdk/pysubmarine/example/pytorch/run_deepfm.sh
##########
@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+
+export JAVA_HOME=${JAVA_HOME:-$HOME/workspace/app/java}
+export HADOOP_HOME=${HADOOP_HOME:-$HADOOP_HDFS_HOME}
+export CLASSPATH=${CLASSPATH:-`hdfs classpath --glob`}
+export ARROW_LIBHDFS_DIR=${ARROW_LIBHDFS_DIR:-$HADOOP_HOME/lib/native}
+
+# path to pysubmarine/submarine
+PYTHONPATH=$HOME/workspace/submarine/submarine-sdk/pysubmarine
+
+HADOOP_CONF_PATH=${HADOOP_CONF_PATH:-$HADOOP_CONF_DIR}
+
+SUBMARINE_VERSION=0.4.0-SNAPSHOT
+SUBMARINE_HADOOP_VERSION=2.9
+SUBMARINE_JAR=/opt/submarine-dist-${SUBMARINE_VERSION}-hadoop-${SUBMARINE_HADOOP_VERSION}/submarine-dist-${SUBMARINE_VERSION}-hadoop-${SUBMARINE_HADOOP_VERSION}/submarine-all-${SUBMARINE_VERSION}-hadoop-${SUBMARINE_HADOOP_VERSION}.jar
+
+java -cp $(${HADOOP_COMMON_HOME}/bin/hadoop classpath --glob):${SUBMARINE_JAR}:${HADOOP_CONF_PATH} \
+ org.apache.submarine.client.cli.Cli job run --name deepfm-job-001 \
+ --framework pytorch \
+ --verbose \
+ --input_path "" \
+ --num_workers 2 \
+ --worker_resources memory=1G,vcores=1 \
+ --worker_launch_cmd "JAVA_HOME=$JAVA_HOME HADOOP_HOME=$HADOOP_HOME CLASSPATH=$CLASSPATH ARROW_LIBHDFS_DIR=$ARROW_LIBHDFS_DIR PYTHONPATH=$PYTHONPATH sdk.zip/sdk/bin/python run_ctr.py --conf ./deepfm.json --task_type train" \
+ --insecure \
+ --conf tony.containers.resources=sdk.zip#archive,${SUBMARINE_JAR},run_ctr.py,deepfm.json
+
+
+
+

Review comment:
       Should remove redundant lines




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org