orchestrator_multimno

Module that orchestrates MultiMNO pipeline components. A spark-submit will be performed for each component in the pipeline.

Usage:

python multimno/orchestrator.py <pipeline.json>

• pipeline.json: Path to a json file with the pipeline configuration.

create_logger(general_config_path)

Create a logger with the specified configuration.

Parameters:

Name	Type	Description	Default
general_config_path	str	The path to the general configuration file.	required

Returns:

Type	Description
	logging.Logger: The created logger object.

Source code in multimno/orchestrator_multimno.py

def create_logger(general_config_path: str):
    """
    Create a logger with the specified configuration.

    Args:
        general_config_path (str): The path to the general configuration file.

    Returns:
        logging.Logger: The created logger object.
    """
    # Get log configuration
    config = parse_configuration(general_config_path)

    report_path = config.get(LoggerKeys.LOG_CONFIG_KEY, LoggerKeys.REPORT_PATH, fallback=None)
    file_format = config.get(LoggerKeys.LOG_CONFIG_KEY, LoggerKeys.FILE_FORMAT, fallback=None)
    datefmt = config.get(LoggerKeys.LOG_CONFIG_KEY, LoggerKeys.DATEFMT, fallback=None)

    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)  # Set the logging level

    # Get log path
    today = datetime.now().strftime("%y%m%d")
    log_path = f"{report_path}/multimno_{today}.log"
    # Make report path + log dir
    os.makedirs(os.path.dirname(log_path), exist_ok=True)

    # Create a file handler
    file_handler = logging.FileHandler(log_path)
    file_handler.setLevel(logging.DEBUG)

    # Create a console handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.DEBUG)

    # Create a formatter and add it to the handlers
    formatter = logging.Formatter(fmt=file_format, datefmt=datefmt)
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)

    # Add the handlers to the logger
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

    return logger