如何将带有argparse的AWS Sagemaker SM_USER_ARGS解析为argparse命名空间？

问题描述 投票：0回答：1

AWS Sagemaker使用SM_USER_ARGS（已记录here）作为环境变量，其中包含用户传递的自变量的字符串（列表）。因此，环境变量值如下所示：'["--test_size","0.2","--random_seed","42", "--not_optmize"]'。

使用json.loads()，我能够将该字符串转换为python列表。虽然，我想创建一个抽象模块，以返回argparse命名空间的方式，无论我在本地还是在AWS Sagemaker服务中运行其余代码，该代码都保持不变。

所以，基本上，我想要的是一个接收输入["--test_size","0.2","--random_seed","42", "--not_optmize"]和输出Namespace(test_size=0.2, random_seed='42', not_optmize=True, <other_arguments>... ])的代码。

python argparse包可以帮助我吗？我正在尝试找出不需要重新实现argparse解析器的方法。

这里是一个例子，我有这个config.ini文件：

[Docker]
home_dir = /opt
SM_MODEL_DIR = %(home_dir)s/ml/model
SM_CHANNELS = ["training"]
SM_NUM_GPUS = 1
SM_NUM_CPUS =
SM_LOG_LEVEL = 20
SM_USER_ARGS = ["--test_size","0.2","--random_seed","42"]
SM_INPUT_DIR = %(home_dir)s/ml/input
SM_INPUT_CONFIG_DIR = %(home_dir)s/ml/input/config
SM_OUTPUT_DIR = %(home_dir)s/ml/output
SM_OUTPUT_INTERMEDIATE_DIR = %(home_dir)s/ml/output/intermediate

我有这个Argparser类：

import argparse
import configparser
import datetime
import json
import multiprocessing
import os
import time
from pathlib import Path
from typing import Any, Dict

from .files import JsonFile, YAMLFile


class ArgParser(ABC):

    @abstractmethod
    def get_arguments(self) -> Dict[str, Any]:
        pass


class AWSArgParser(ArgParser):

    def __init__(self):
        configuration_file_path = 'config.ini'

        self.environment = "Sagemaker" \
            if os.environ.get("SM_MODEL_DIR", False) \
            else os.environ.get("ENVIRON", "Default")

        config = configparser.ConfigParser()
        config.read(configuration_file_path)
        if self.environment == "Local":
            config[self.environment]["home_dir"] = str(pathlib.Path(__file__).parent.absolute())
        if self.environment != 'Sagemaker':
            config[self.environment]["SM_NUM_CPUS"] = str(multiprocessing.cpu_count())

        for key, value in config[self.environment].items():
            os.environ[key.upper()] = value

        self.parser = argparse.ArgumentParser()
        # AWS Sagemaker default environmental arguments
        self.parser.add_argument(
            '--model_dir',
            type=str,
            default=os.environ['SM_MODEL_DIR'],
        )
        self.parser.add_argument(
            '--channel_names',
            default=json.loads(os.environ['SM_CHANNELS']),
        )
        self.parser.add_argument(
            '--num_gpus',
            type=int,
            default=os.environ['SM_NUM_GPUS'],
        )
        self.parser.add_argument(
            '--num_cpus',
            type=int,
            default=os.environ['SM_NUM_CPUS'],
        )
        self.parser.add_argument(
            '--user_args',
            default=json.loads(os.environ['SM_USER_ARGS']),
        )
        self.parser.add_argument(
            '--input_dir',
            type=str,
            default=os.environ['SM_INPUT_DIR'],
        )
        self.parser.add_argument(
            '--input_config_dir',
            type=Path,
            default=os.environ['SM_INPUT_CONFIG_DIR'],
        )
        self.parser.add_argument(
            '--output_dir',
            type=Path,
            default=os.environ['SM_OUTPUT_DIR'],
        )

        # Extra arguments
        self.run_tag = datetime.datetime \
            .fromtimestamp(time.time()) \
            .strftime('%Y-%m-%d-%H-%M-%S')
        self.parser.add_argument(
            '--run_tag',
            default=self.run_tag,
            type=str,
            help=f"Run tag (default: 'datetime.fromtimestamp')",
        )
        self.parser.add_argument(
            '--environment',
            type=str,
            default=self.environment,
        )

        self.args = self.parser.parse_args()

    def get_arguments(self) -> Dict[str, Any]:
        <parse self.args.user_args>

        return self.args

然后我有了我的train脚本：

from utils.arg_parser import AWSArgParser

if __name__ == '__main__':
    logger.info(f"Begin train.py")

    if os.environ["ENVIRON"] == "Sagemaker":
        arg_parser = AWSArgParser()
        args = arg_parser.get_arguments()
    else:
        args = <normal local parse>
python amazon-web-services argparse amazon-sagemaker
1个回答
0
投票

在@chepner的注释之后，示例解决方案将是这样的：

config.ini文件：

[Docker]
home_dir = /opt
SM_MODEL_DIR = %(home_dir)s/ml/model
SM_CHANNELS = ["training"]
SM_NUM_GPUS = 1
SM_NUM_CPUS =
SM_LOG_LEVEL = 20
SM_USER_ARGS = ["--test_size","0.2","--random_seed","42", "--not_optimize"]
SM_INPUT_DIR = %(home_dir)s/ml/input
SM_INPUT_CONFIG_DIR = %(home_dir)s/ml/input/config
SM_OUTPUT_DIR = %(home_dir)s/ml/output
SM_OUTPUT_INTERMEDIATE_DIR = %(home_dir)s/ml/output/intermediate

TrainArgParser类，例如：

class ArgParser(ABC):

    @abstractmethod
    def get_arguments(self) -> Dict[str, Any]:
        pass


class TrainArgParser(ArgParser):

    def __init__(self):
        configuration_file_path = 'config.ini'

        self.environment = "Sagemaker" \
            if os.environ.get("SM_MODEL_DIR", False) \
            else os.environ.get("ENVIRON", "Default")

        config = configparser.ConfigParser()
        config.read(configuration_file_path)
        if self.environment == "Local":
            config[self.environment]["home_dir"] = str(pathlib.Path(__file__).parent.absolute())
        if self.environment != 'Sagemaker':
            config[self.environment]["SM_NUM_CPUS"] = str(multiprocessing.cpu_count())

        for key, value in config[self.environment].items():
            os.environ[key.upper()] = value

        self.parser = argparse.ArgumentParser()
        # AWS Sagemaker default environmental arguments
        self.parser.add_argument(
            '--model_dir',
            type=str,
            default=os.environ['SM_MODEL_DIR'],
        )
        self.parser.add_argument(
            '--channel_names',
            default=json.loads(os.environ['SM_CHANNELS']),
        )
        self.parser.add_argument(
            '--num_gpus',
            type=int,
            default=os.environ['SM_NUM_GPUS'],
        )
        self.parser.add_argument(
            '--num_cpus',
            type=int,
            default=os.environ['SM_NUM_CPUS'],
        )
        self.parser.add_argument(
            '--user_args',
            default=json.loads(os.environ['SM_USER_ARGS']),
        )
        self.parser.add_argument(
            '--input_dir',
            type=str,
            default=os.environ['SM_INPUT_DIR'],
        )
        self.parser.add_argument(
            '--input_config_dir',
            type=Path,
            default=os.environ['SM_INPUT_CONFIG_DIR'],
        )
        self.parser.add_argument(
            '--output_dir',
            type=Path,
            default=os.environ['SM_OUTPUT_DIR'],
        )

        # Extra arguments
        self.run_tag = datetime.datetime \
            .fromtimestamp(time.time()) \
            .strftime('%Y-%m-%d-%H-%M-%S')
        self.parser.add_argument(
            '--run_tag',
            default=self.run_tag,
            type=str,
            help=f"Run tag (default: 'datetime.fromtimestamp')",
        )
        self.parser.add_argument(
            '--environment',
            type=str,
            default=self.environment,
        )

        self.args = self.parser.parse_args()

    def get_arguments(self) -> Dict[str, Any]:
        # Not in AWS Sagemaker arguments
        self.parser.add_argument(
            '--test_size',
            default=0.2,
            type=float,
            help="Test dataset size (default: '0.2')",
        )
        self.parser.add_argument(
            '--random_seed',
            default=42,
            type=int,
            help="Random number for initialization (default: '42')",
        )
        self.parser.add_argument(
            '--secrets',
            type=YAMLFile.parse_string,
            default='',
            help="An yaml formated string (default: '')"
        )
        self.parser.add_argument(
            '--bucket_name',
            type=str,
            default='',
            help="Bucket name of a remote storage (default: '')"
        )
        self.args = self.parser.parse_args(self.args.user_args)

        return self.args

和train的entry_script会像这样开始：

#!/usr/bin/env python

from utils.arg_parser import TrainArgParser

if __name__ == '__main__':
    logger.info(f"Begin train.py")

    arg_parser = TrainArgParser()
    args = arg_parser.get_arguments()
    print(args)

这应该输出如下内容：

Namespace(bucket_name='', channel_names=['training'], environment='Docker', input_config_dir=PosixPath('/opt/ml/input/config'), input_dir='/opt/ml/input', model_dir='/opt/ml/model', num_cpus=8, num_gpus=1, output_dir=PosixPath('/opt/ml/output'), random_seed=42, run_tag='2020-03-11-22-18-21', secrets={}, test_size=0.2, user_args=['--test_size', '0.2', '--random_seed', '42'])

但是如果AWS Sagemaker将SM_USER_ARGS和SM_HPS视为同一事物，那就没用了。 ：（

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.