This is an automated email from the ASF dual-hosted git repository.
byronhsu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/submarine.git
The following commit(s) were added to refs/heads/master by this push:
new 56bfbbb SUBMARINE-858. Implement log_param and log_metric for
distributed scenarios
56bfbbb is described below
commit 56bfbbbc0f6f0eaedd4c06fb0ad2d8625c3af501
Author: ByronHsu <[email protected]>
AuthorDate: Mon Jun 21 11:58:39 2021 +0800
SUBMARINE-858. Implement log_param and log_metric for distributed scenarios
### What is this PR for?
Implement the backbone of distributed model management sdk and provide an
example.
**Implement details:**
1. Calling start(), ModelsClient will parse the worker_id + experiment_id
and creates a Mlflow run.
2. The Mlflow run name is worker_id, and the Mflow experiment name is
experiment_id.
3. Calling log_param() or log_metric(), their info will be directed to the
Mlflow runs previously set.
**I also wrote an example docker image to test and demo the function.**
You can follow [this
instruction](https://github.com/apache/submarine/blob/67a595a8c7122d8026f151e5bb49292b7df3ef51/dev-support/examples/tracking/readme.md)
to try this new function. (**REMEMBER TO INSTALL LATEST PYSUBMARINE LOCALLY
FIRST**)
### What type of PR is it?
[Feature]
### Todos
* [ ] - Task
### What is the Jira issue?
https://issues.apache.org/jira/browse/SUBMARINE-858
### How should this be tested?
Manually test by the example provided.
### Screenshots (if appropriate)
https://www.loom.com/share/bc0e9db28a0243bdacb564aacef6306b
### Questions:
* Do the license files need updating? No
* Are there breaking changes for older versions? No
* Does this need new documentation? No
Author: ByronHsu <[email protected]>
Author: Byron <[email protected]>
Signed-off-by: byronhsu <[email protected]>
Closes #609 from ByronHsu/distributed_mm and squashes the following commits:
2fa03bb9 [ByronHsu] fix typo
42deeb70 [ByronHsu] add license
a3543565 [ByronHsu] lint fails
59f44298 [ByronHsu] uncomment script
82869057 [ByronHsu] add example readme
b2dcfd06 [ByronHsu] remove tmp test
dce819b6 [ByronHsu] add tracking example
2205ea08 [Byron] distributed mm initial commit
---
.github/workflows/master.yml | 2 +-
.gitignore | 6 +-
dev-support/examples/tracking/Dockerfile | 22 +++++
dev-support/examples/tracking/build.sh | 44 ++++++++++
dev-support/examples/tracking/post.sh | 42 ++++++++++
dev-support/examples/tracking/readme.md | 22 +++++
.../examples/tracking/tracking.py | 13 +++
.../pysubmarine/submarine/models/client.py | 45 ++++++++++-
.../pysubmarine/submarine/models/utils.py | 94 ++++++++++++++++++----
9 files changed, 269 insertions(+), 21 deletions(-)
diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml
index 24af976..2fed76f 100644
--- a/.github/workflows/master.yml
+++ b/.github/workflows/master.yml
@@ -22,7 +22,7 @@ jobs:
java -version
- uses: actions/cache@v1
with:
- path: ~/.m2/repository/com1
+ path: ~/.m2/repository/com
key: ${{ runner.os }}-maven-com-${{ hashFiles('**/pom.xml') }}
restore-keys: |
${{ runner.os }}-maven-com-
diff --git a/.gitignore b/.gitignore
index 1bd36f5..04fb86b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -102,4 +102,8 @@ submarine-cloud-v2/charts/*
.ipynb_checkpoints
# redundant logging files
-*.log
\ No newline at end of file
+*.log
+
+# mlflow
+mlruns/
+mlflow*
diff --git a/dev-support/examples/tracking/Dockerfile
b/dev-support/examples/tracking/Dockerfile
new file mode 100644
index 0000000..d6148af
--- /dev/null
+++ b/dev-support/examples/tracking/Dockerfile
@@ -0,0 +1,22 @@
+# 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 python:3.7
+MAINTAINER Apache Software Foundation <[email protected]>
+
+ADD ./tmp/submarine-sdk /opt/
+RUN pip install /opt/pysubmarine
+
+ADD ./tracking.py /opt/
\ No newline at end of file
diff --git a/dev-support/examples/tracking/build.sh
b/dev-support/examples/tracking/build.sh
new file mode 100755
index 0000000..b83f0e8
--- /dev/null
+++ b/dev-support/examples/tracking/build.sh
@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+# 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.
+
+set -euxo pipefail
+
+SUBMARINE_VERSION=0.6.0-SNAPSHOT
+SUBMARINE_IMAGE_NAME="tracking:${SUBMARINE_VERSION}"
+
+if [ -L ${BASH_SOURCE-$0} ]; then
+ PWD=$(dirname $(readlink "${BASH_SOURCE-$0}"))
+else
+ PWD=$(dirname ${BASH_SOURCE-$0})
+fi
+export CURRENT_PATH=$(cd "${PWD}">/dev/null; pwd)
+export SUBMARINE_HOME=${CURRENT_PATH}/../../..
+
+if [ -d "${CURRENT_PATH}/tmp" ] # if old tmp folder is still there, delete it.
+then
+ rm -rf "${CURRENT_PATH}/tmp"
+fi
+
+mkdir -p "${CURRENT_PATH}/tmp"
+cp -r "${SUBMARINE_HOME}/submarine-sdk" "${CURRENT_PATH}/tmp"
+
+# build image
+cd ${CURRENT_PATH}
+echo "Start building the ${SUBMARINE_IMAGE_NAME} docker image ..."
+docker build -t ${SUBMARINE_IMAGE_NAME} .
+
+# clean temp file
+rm -rf "${CURRENT_PATH}/tmp"
diff --git a/dev-support/examples/tracking/post.sh
b/dev-support/examples/tracking/post.sh
new file mode 100755
index 0000000..7adbd7d
--- /dev/null
+++ b/dev-support/examples/tracking/post.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# 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.
+
+curl -X POST -H "Content-Type: application/json" -d '
+{
+ "meta": {
+ "name": "tracking-example",
+ "namespace": "default",
+ "framework": "TensorFlow",
+ "cmd": "python /opt/tracking.py",
+ "envVars": {
+ "ENV_1": "ENV1"
+ }
+ },
+ "environment": {
+ "image": "tracking:0.6.0-SNAPSHOT"
+ },
+ "spec": {
+ "Ps": {
+ "replicas": 1,
+ "resources": "cpu=1,memory=128M"
+ },
+ "Worker": {
+ "replicas": 3,
+ "resources": "cpu=1,memory=128M"
+ }
+ }
+}
+' http://127.0.0.1:32080/api/v1/experiment
\ No newline at end of file
diff --git a/dev-support/examples/tracking/readme.md
b/dev-support/examples/tracking/readme.md
new file mode 100644
index 0000000..e452a16
--- /dev/null
+++ b/dev-support/examples/tracking/readme.md
@@ -0,0 +1,22 @@
+# Tracking Example
+
+## Usage
+This is an easy example of how to track the metric and paramater in
submarine-sdk. Basically, the sdk will detect which experiment and worker-id
are you at, and log your data to the corresponding place.
+
+For example, you start an experiment with 3 workers. Suppose the experiment is
assigned with an ID `experiment_12345678`, and the operator launches 3 pods
with worker_id `worker-0`, `worker-1` and `worker-2` respectively.
+
+The logging of `worker-i` will be directed to `experiment_12345678` /
`worker-i` in the mlflow server
+
+## How to execute
+
+1. Build the docker image
+
+```bash
+./dev-support/examples/tracking/build.sh
+```
+
+2. Submit a post request
+
+```bash
+./dev-support/examples/tracking/post.sh
+```
diff --git a/submarine-sdk/pysubmarine/submarine/models/utils.py
b/dev-support/examples/tracking/tracking.py
similarity index 67%
copy from submarine-sdk/pysubmarine/submarine/models/utils.py
copy to dev-support/examples/tracking/tracking.py
index 273b024..641c25e 100644
--- a/submarine-sdk/pysubmarine/submarine/models/utils.py
+++ b/dev-support/examples/tracking/tracking.py
@@ -14,3 +14,16 @@
specific language governing permissions and limitations
under the License.
"""
+
+from submarine import ModelsClient
+import random
+import time
+
+if __name__ == "__main__":
+ periscope = ModelsClient()
+ with periscope.start() as run:
+ periscope.log_param("learning_rate", random.random())
+ for i in range(100):
+ time.sleep(1)
+ periscope.log_metric("mse", random.random() * 100, i)
+ periscope.log_metric("acc", random.random(), i)
\ No newline at end of file
diff --git a/submarine-sdk/pysubmarine/submarine/models/client.py
b/submarine-sdk/pysubmarine/submarine/models/client.py
index 7bf6495..1bd7f22 100644
--- a/submarine-sdk/pysubmarine/submarine/models/client.py
+++ b/submarine-sdk/pysubmarine/submarine/models/client.py
@@ -17,10 +17,12 @@
import os
import mlflow
+from mlflow.exceptions import MlflowException
from mlflow.tracking import MlflowClient
from .constant import (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,
MLFLOW_S3_ENDPOINT_URL, MLFLOW_TRACKING_URI)
+from .utils import get_job_id, get_worker_index
class ModelsClient():
@@ -29,12 +31,34 @@ class ModelsClient():
"""
Set up mlflow server connection, including: s3 endpoint, aws, tracking
server
"""
+ # if setting url in environment variable,
+ # there is no need to set it by MlflowClient() or
mlflow.set_tracking_uri() again
os.environ[
"MLFLOW_S3_ENDPOINT_URL"] = registry_uri or MLFLOW_S3_ENDPOINT_URL
os.environ["AWS_ACCESS_KEY_ID"] = AWS_ACCESS_KEY_ID
os.environ["AWS_SECRET_ACCESS_KEY"] = AWS_SECRET_ACCESS_KEY
os.environ["MLFLOW_TRACKING_URI"] = tracking_uri or MLFLOW_TRACKING_URI
- self._client = MlflowClient()
+ self.client = MlflowClient()
+
+ def start(self):
+ """
+ 1. Start a new Mlflow run
+ 2. Direct the logging of the artifacts and metadata
+ to the Run named "worker_i" under Experiment "job_id"
+ 3. If in distributed training, worker and job id would be parsed from
environment variable
+ 4. If in local traning, worker and job id will be generated.
+ :return: Active Run
+ """
+ experiment_name = get_job_id()
+ run_name = get_worker_index()
+ experiment_id = self._get_or_create_experiment(experiment_name)
+ return mlflow.start_run(run_name=run_name, experiment_id=experiment_id)
+
+ def log_param(self, key, value):
+ mlflow.log_param(key, value)
+
+ def log_metric(self, key, value, step=None):
+ mlflow.log_metric(key, value, step)
def log_model(self, name, checkpoint):
mlflow.pytorch.log_model(registered_model_name=name,
@@ -46,7 +70,22 @@ class ModelsClient():
return model
def update_model(self, name, new_name):
- self._client.rename_registered_model(name=name, new_name=new_name)
+ self.client.rename_registered_model(name=name, new_name=new_name)
def delete_model(self, name, version):
- self._client.delete_model_version(name=name, version=version)
+ self.client.delete_model_version(name=name, version=version)
+
+ def _get_or_create_experiment(self, experiment_name):
+ """
+ Return the id of experiment.
+ If non-exist, create one. Otherwise, return the existing one.
+ :return: Experiment id
+ """
+ try:
+ experiment = mlflow.get_experiment_by_name(experiment_name)
+ if experiment is None: # if not found
+ raise MlflowException("No valid experiment has been found")
+ return experiment.experiment_id # if found
+ except MlflowException:
+ experiment = mlflow.create_experiment(name=experiment_name)
+ return experiment
diff --git a/submarine-sdk/pysubmarine/submarine/models/utils.py
b/submarine-sdk/pysubmarine/submarine/models/utils.py
index 273b024..7bb5d5d 100644
--- a/submarine-sdk/pysubmarine/submarine/models/utils.py
+++ b/submarine-sdk/pysubmarine/submarine/models/utils.py
@@ -1,16 +1,78 @@
-"""
- 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.
-"""
+# 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 __future__ import print_function
+
+import json
+import os
+import uuid
+
+from submarine.utils import env
+
+_JOB_ID_ENV_VAR = "JOB_ID"
+
+_TF_CONFIG = "TF_CONFIG"
+_CLUSTER_SPEC = "CLUSTER_SPEC"
+_JOB_NAME = "JOB_NAME"
+_TYPE = "type"
+_TASK = "task"
+_INDEX = "index"
+_RANK = "RANK"
+
+
+def get_job_id():
+ """
+ Get the current experiment id.
+ :return The experiment id:
+ """
+ # Get yarn application or K8s experiment ID when running distributed
training
+ if env.get_env(_JOB_ID_ENV_VAR) is not None:
+ return env.get_env(_JOB_ID_ENV_VAR)
+ else: # set Random ID when running local training
+ job_id = uuid.uuid4().hex
+ os.environ[_JOB_ID_ENV_VAR] = job_id
+ return job_id
+
+
+def get_worker_index():
+ """
+ Get the current worker index.
+ :return: The worker index:
+ """
+ # Get TensorFlow worker index
+ if env.get_env(_TF_CONFIG) is not None:
+ tf_config = json.loads(os.environ.get(_TF_CONFIG))
+ task_config = tf_config.get(_TASK)
+ task_type = task_config.get(_TYPE)
+ task_index = task_config.get(_INDEX)
+ worker_index = task_type + '-' + str(task_index)
+ elif env.get_env(_CLUSTER_SPEC) is not None:
+ cluster_spec = json.loads(os.environ.get(_CLUSTER_SPEC))
+ task_config = cluster_spec.get(_TASK)
+ task_type = task_config.get(_JOB_NAME)
+ task_index = task_config.get(_INDEX)
+ worker_index = task_type + '-' + str(task_index)
+ # Get PyTorch worker index
+ elif env.get_env(_RANK) is not None:
+ rank = env.get_env(_RANK)
+ if rank == "0":
+ worker_index = "master-0"
+ else:
+ worker_index = "worker-" + rank
+ # Set worker index to "worker-0" When running local training
+ else:
+ worker_index = "worker-0"
+
+ return worker_index
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
