feat: Python SDK for Kubeflow Training Operator #17
Comments
PyTorchJobClient
User can loads authentication and cluster information from kube-config file and stores them in kubernetes.client.configuration. Parameters are as following:
The APIs for PyTorchJobClient are as following:
create
Create the provided pytorchjob in the specified namespace Examplefrom kubernetes.client import V1PodTemplateSpec
from kubernetes.client import V1ObjectMeta
from kubernetes.client import V1PodSpec
from kubernetes.client import V1Container
from kubernetes.client import V1ResourceRequirements
from kubeflow.training import constants
from kubeflow.training import utils
from kubeflow.training import V1ReplicaSpec
from kubeflow.training import KubeflowOrgV1PyTorchJob
from kubeflow.training import KubeflowOrgV1PyTorchJobSpec
from kubeflow.training import PyTorchJobClient
container = V1Container(
name="pytorch",
image="gcr.io/kubeflow-ci/pytorch-dist-mnist-test:v1.0",
args=["--backend","gloo"],
)
master = V1ReplicaSpec(
replicas=1,
restart_policy="OnFailure",
template=V1PodTemplateSpec(
spec=V1PodSpec(
containers=[container]
)
)
)
worker = V1ReplicaSpec(
replicas=1,
restart_policy="OnFailure",
template=V1PodTemplateSpec(
spec=V1PodSpec(
containers=[container]
)
)
)
pytorchjob = KubeflowOrgV1PyTorchJob(
api_version="kubeflow.org/v1",
kind="PyTorchJob",
metadata=V1ObjectMeta(name="mnist", namespace='default'),
spec=KubeflowOrgV1PyTorchJobSpec(
clean_pod_policy="None",
pytorch_replica_specs={"Master": master,
"Worker": worker}
)
)
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.create(pytorchjob)Parameters
Return typeobject get
Get the created pytorchjob in the specified namespace Examplefrom kubeflow.training import pytorchjobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.get('mnist', namespace='kubeflow')Parameters
Return typeobject patch
Patch the created pytorchjob in the specified namespace. Note that if you want to set the field from existing value to Examplepytorchjob = KubeflowOrgV1PyTorchJob(
api_version="kubeflow.org/v1",
... #update something in PyTorchJob spec
)
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.patch('mnist', isvc)Parameters
Return typeobject delete
Delete the created pytorchjob in the specified namespace Examplefrom kubeflow.training import pytorchjobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.delete('mnist', namespace='kubeflow')Parameters
Return typeobject wait_for_job
Wait for the specified job to finish. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.wait_for_job('mnist', namespace='kubeflow')
# The API also supports watching the PyTorchJob status till it's Succeeded or Failed.
pytorchjob_client.wait_for_job('mnist', namespace='kubeflow', watch=True)
NAME STATE TIME
pytorch-dist-mnist-gloo Created 2020-01-02T09:21:22Z
pytorch-dist-mnist-gloo Running 2020-01-02T09:21:36Z
pytorch-dist-mnist-gloo Running 2020-01-02T09:21:36Z
pytorch-dist-mnist-gloo Running 2020-01-02T09:21:36Z
pytorch-dist-mnist-gloo Running 2020-01-02T09:21:36Z
pytorch-dist-mnist-gloo Succeeded 2020-01-02T09:26:38ZParameters
Return typeobject wait_for_condition
Waits until any of the specified conditions occur. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.wait_for_condition('mnist', expected_condition=["Succeeded", "Failed"], namespace='kubeflow')Parameters
Return typeobject get_job_status
Returns PyTorchJob status, such as Running, Failed or Succeeded. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.get_job_status('mnist', namespace='kubeflow')Parameters
Return typeStr is_job_running
Returns True if the PyTorchJob running; false otherwise. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.is_job_running('mnist', namespace='kubeflow')Parameters
Return typeBool is_job_succeeded
Returns True if the PyTorchJob succeeded; false otherwise. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.is_job_succeeded('mnist', namespace='kubeflow')Parameters
Return typeBool get_pod_names
Get pod names of the PyTorchJob. Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.get_pod_names('mnist', namespace='kubeflow')Parameters
Return typeSet get_logs
Get training logs of the PyTorchJob. By default only get the logs of Pod that has labels 'job-role: master', to get all pods logs, specfy the Examplefrom kubeflow.training import PyTorchJobClient
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.get_logs('mnist', namespace='kubeflow')Parameters
Return typeStr |
Test result for PyTorchJob using Python SDK for Kubeflow# Install Kubeflow training package for Python
!pip install kubeflow-training# import python client packages for kubernetes
from kubernetes.client import V1PodTemplateSpec
from kubernetes.client import V1ObjectMeta
from kubernetes.client import V1PodSpec
from kubernetes.client import V1Container
from kubernetes.client import V1ResourceRequirements
# import python client packages for kubeflow
from kubeflow.training import constants
from kubeflow.training.utils import utils
from kubeflow.training import V1ReplicaSpec
from kubeflow.training import V1PyTorchJob
from kubeflow.training import V1PyTorchJobSpec
from kubeflow.training import V1RunPolicy
from kubeflow.training import PyTorchJobClientnamespace = utils.get_current_k8s_namespace()
print('# Current namespace: ', namespace)# For TrainDB-ML, we need to prepare the following elements and parameters.
# 1. TrainingOperator Type: e.g. name="pytorch"
# 2. Prebuild docker images in image repository (e.g., dockerhub)
# - RSPN, MDN, Autoregressive Model, HyperSPN, etc.
# 3. Parameters and hyperparameters for training
# - P#1. Model name
# - P#2. Database access info. (dbuser/dbpassword)
# - P#3. Table info. ([optional] schema info.)
# - P#4.
# MNIST test container
container = V1Container(
name="pytorch",
image="gcr.io/kubeflow-ci/pytorch-dist-mnist-test:v1.0",
args=["--backend","gloo"]
)# Master node specification
master = V1ReplicaSpec(
replicas=1,
restart_policy="OnFailure",
template=V1PodTemplateSpec(
spec=V1PodSpec(
containers=[container]
)
)
)# Worker node specification
worker = V1ReplicaSpec(
replicas=1,
restart_policy="OnFailure",
template=V1PodTemplateSpec(
spec=V1PodSpec(
containers=[container]
)
)
)# PyTorchJob Definition
pytorchjob = V1PyTorchJob(
api_version="kubeflow.org/v1",
kind="PyTorchJob",
metadata=V1ObjectMeta(name="pytorch-dist-mnist-gloo",namespace=namespace),
spec=V1PyTorchJobSpec(
run_policy=V1RunPolicy(clean_pod_policy="None"),
pytorch_replica_specs={"Master": master,
"Worker": worker}
)
)# Create PyTorchJob Client
pytorchjob_client = PyTorchJobClient()
pytorchjob_client.create(pytorchjob)# Get depolyment information for PyTorchJob
pytorchjob_client.get('pytorch-dist-mnist-gloo')# Querying PyTorchJob Client Status
pytorchjob_client.get_job_status('pytorch-dist-mnist-gloo', namespace=namespace)pytorchjob_client.wait_for_job('pytorch-dist-mnist-gloo', namespace=namespace, watch=True)pytorchjob_client.is_job_succeeded('pytorch-dist-mnist-gloo', namespace=namespace)pytorchjob_client.get_logs('pytorch-dist-mnist-gloo', namespace=namespace)pytorchjob_client.delete('pytorch-dist-mnist-gloo') |
Java에서 Python 사용하기현실적인 방안자바에서 파이썬을 사용하는 현실적인 방안은 두 가지로 볼 수 있다.
데몬을 만들어서 서비스 하는 방식파이썬으로 서버를 만들어서 자바에서 서비스를 호출하는 방식인데 운영환경에서 너무 많은 서버를 두어야 해서 일단 이 방식은 사용하지 않기로 한다. 프로세스를 사용하는 방식자바에서 파이썬 프로그램을 호출하는 방식인데 java.lang.Runtime이나 java.lang.ProcessBuilder를 사용하면 시스템의 프로그램을 실행하고 그 결과를 받아 올 수 있다. 보다 자세한 설명은 https://www.baeldung.com/java-lang-processbuilder-api 을 참고한다. 즉, 자바에서 파이썬 프로그램을 실행하고 파이썬이 표준 출력(sdtout)에 쓴 내용을 문자열로 캡처하여 처리하는 방식으로 개발한다. 예를들어, 이미지를 텍스트로 변환하는 프로그램을 파이썬으로 작성한다.
ProcessBuilder의 기본 사용방법ProcessBuilder를 생성할 때 명령어와 인자를 전달한다. 인스턴스의 start() 메소드를 실행하고 start 메소드가 반환하는 Process 인스턴스의 watiFor() 메소드를 사용하여 서브 프로세스가 끝날 때까지 기다린다. waitFor() 메소드는 정수값을 반환하는데 보통 0이면 정상이다. String command = "python"; // 명령어
String arg1 = "mnist.py"; // 인자
ProcessBuilder builder = new ProcessBuilder(command, arg1);
Process process = builder.start();
int exitVal = process.waitFor(); // 자식 프로세스가 종료될 때까지 기다림
if(exitVal != 0) {
// 비정상 종료
}위 코드에서는 python.exe를 명령어로 전달하고 실행할 파이썬 프로그램을 인자로 전달했다. 파이썬이 파이썬 프로그램을 실행할 것이다. 자식 프로세스가 표준출력(System.out)으로 출력하는 것을 가져오려면 process.getInputStream()을 사용한다. InputStream input = process.getInputStream(); // 자식 프로세스가 System.out에 출력하는 내용 자식 프로세스에게 입력을 전달하려면 process.getOutputStream()을 사용한다. OutputStream output = process.getOutputStream(); //자식 프로세스에 입력값 전달파이썬 파일을 실행하여 파이썬 파일에서 System.out으로 출력하는 내용을 가져와서 출력하는 완전한 코드는 다음과 같다. String command = "C:\\Anaconda3\\envs\\jep\\python.exe"; // 명령어
String arg1 = "F:\\src\\hyeon\\latteonterrace\\python\\python-exe\\src\\python\\test.py"; // 인자
ProcessBuilder builder = new ProcessBuilder(command, arg1);
Process process = builder.start();
int exitVal = process.waitFor(); // 자식 프로세스가 종료될 때까지 기다림
BufferedReader br = new BufferedReader(new InputStreamReader(process.getInputStream(), "euc-kr")); // 서브 프로세스가 출력하는 내용을 받기 위해
String line;
while ((line = br.readLine()) != null) {
System.out.println(">>> " + line); // 표준출력에 쓴다
}
if(exitVal != 0) {
// 비정상 종료
System.out.println("서브 프로세스가 비정상 종료되었다.");
}
파이썬에서 파일 쓰기를 하거나 크롤링을 할 때 한글이 깨지는 문제가 있다. 파이썬이 기본 UTF-8이 아니라서 비 영어권 사용자들은 코딩할 때 별도의 옵션을 제공해야 한다. 인코딩 타입을 지정하면 해결할 수 있다. file=open(fileName, 'w', encoding='utf-8')크롤링하다가 다음과 같은 에러를 만날 수도 있다. UnicodeEncodeError: 'cp949' codec can't encode character ...코드 상단에 다음과 같이 코드를 추가하면 해결할 수 있다. import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8")
sys.stderr = io.TextIOWrapper(sys.stderr.detach(), encoding="utf-8")redirectErrorStream(true)ProcessBuilder를 사용하여 서브 프로그램을 실행할 때 어떤 오류가 발생했는지 알 수 없게 된다. redirectErrorStream(true)를 사용하여 표준 에러 출력을 표준 출력으로 쓸 수 있게 할 수 있다. ProcessBuilder builder = new ProcessBuilder(pythonExe, pyFilePath + "/test.py");
builder.redirectErrorStream(true); // 표준 에러도 표준 출력에 쓴다
process = builder.start();
BufferedReader br = new BufferedReader(new InputStreamReader(process.getInputStream(), "utf-8"));이제 서브 프로세스에서 오류가 발생하는 경우 inputStream()을 통해 읽을 수 있다. waitFor()를 사용하여 inputStream을 통해 읽은 내용일 오류인지 정상적인 출력인지 알 수 있다. int exitVal = process.waitFor();
if(exitVal != 0) {
System.out.println("비정상 종료");
}표준 입력과 출력을 Redirecting하기표준입력과 출력을 파일에 쓰기 원할 수 있다. redirectOutput 메소드를 사용하여 파일과 같은 다른 소스에 쓸 수 있다. 이 경우에 getOutputStream()은 ProcessBuilder.NullOutputStream을 반환한다. ProcessBuilder processBuilder = new ProcessBuilder("java", "-version");
processBuilder.redirectErrorStream(true);
File log = folder.newFile("java-version.log");
processBuilder.redirectOutput(log);
Process process = processBuilder.start();표준 출력과 표준 에러를 각각의 파일로 저장할 수 잇다. File outputLog = tempFolder.newFile("standard-output.log");
File errorLog = tempFolder.newFile("error.log");
processBuilder.redirectOutput(Redirect.appendTo(outputLog));
processBuilder.redirectError(Redirect.appendTo(errorLog));현재 프로세스의 I/O를 상속하기redirectOutput() 메서드 등으로 Redirect.INHERIT를 지정하면 부모 프로세스에서 바로 System.out으로 출력한 것처럼 하위 프로세스의 출력이 연결된다. System.out으로 따로 스트림을 복사하지 않아도 콘솔에서 결과가 보인다. 이렇게 Redirect.INHERIT로 지정하면 Process.getInputStream() 메서드의 실행 결과는 java.lang.ProcessBuilder$NullInputStream 클래스가 되며 실제 출력 내용을 스트림으로 전달하지 않는다. redirectOutput() 메서드로 별다른 값을 지정하지 않았을 때의 기본값은 Redirect.PIPE이며 파이프를 통해 부모 프로세스로 출력 결과를 전달한다. 예제 1에서 한 것처럼 기본값일 때는 직접 Process.getInputStream() 메서드로 얻어온 스트림을 다루어야 한다. redirectOutput(File) 메서드로 직접 스트림을 출력할 파일을 지정할 수도 있다. @Test
public void givenProcessBuilder_whenInheritIO_thenSuccess() throws IOException, InterruptedException {
ProcessBuilder processBuilder = new ProcessBuilder("/bin/sh", "-c", "echo hello");
processBuilder.inheritIO();
Process process = processBuilder.start();
int exitCode = process.waitFor();
assertEquals("No errors should be detected", 0, exitCode);
}기타 고려할 사항Process를 사용할 때 몇가지 고려할 사항을 알아보자.
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Python SDK for Kubeflow Training Operator
Training Operator provides Python SDK for the custom resources. More docs are available in sdk/python folder.
Use pip install command to install the latest release of the SDK:
Examples
First, we need to download the examples of Python SDK for Kubeflow Training Operator.
For testing, go to the mnist example folder.
The text was updated successfully, but these errors were encountered: