dmlc--dgl
kylasa
8086d1edde
* Adding launch script and wrapper script to trigger distributed graph partitioning pipeline as defined in the UX document 1. dispatch_data.py is a wrapper script which builds the command and triggers the distributed partitioning pipeline 2. distgraphlaunch.py is the main python script which triggers the pipeline and to simplify its usage dispatch_data.py is included as a wrapper script around it. * Added code to auto-detect python version and retrieve some parameters from the input metadata json file 1. Auto detect python version 2. Read the metadata json file and extract some parameters to pass to the user defined command which is used to trigger the pipeline. * Updated the json file name to metadata.json file per UX documentation 1. Renamed json file name per UX documentation. * address comments * fix * fix doc * use unbuffered logging to cure anxiety * cure more anxiety * Update tools/dispatch_data.py Co-authored-by: Minjie Wang <minjie.wang@nyu.edu> * oops Co-authored-by: Quan Gan <coin2028@hotmail.com> Co-authored-by: Minjie Wang <minjie.wang@nyu.edu>
53 行
2.2 KiB
Python
53 行
2.2 KiB
Python
import argparse
|
|
import numpy as np
|
|
import torch.multiprocessing as mp
|
|
import logging
|
|
import platform
|
|
import os
|
|
from data_shuffle import single_machine_run, multi_machine_run
|
|
|
|
def log_params(params):
|
|
""" Print all the command line arguments for debugging purposes.
|
|
|
|
Parameters:
|
|
-----------
|
|
params: argparse object
|
|
Argument Parser structure listing all the pre-defined parameters
|
|
"""
|
|
print('Input Dir: ', params.input_dir)
|
|
print('Graph Name: ', params.graph_name)
|
|
print('Schema File: ', params.schema)
|
|
print('No. partitions: ', params.num_parts)
|
|
print('Output Dir: ', params.output)
|
|
print('WorldSize: ', params.world_size)
|
|
print('Metis partitions: ', params.partitions_file)
|
|
|
|
if __name__ == "__main__":
|
|
"""
|
|
Start of execution from this point.
|
|
Invoke the appropriate function to begin execution
|
|
"""
|
|
#arguments which are already needed by the existing implementation of convert_partition.py
|
|
parser = argparse.ArgumentParser(description='Construct graph partitions')
|
|
parser.add_argument('--input-dir', required=True, type=str,
|
|
help='The directory path that contains the partition results.')
|
|
parser.add_argument('--graph-name', required=True, type=str,
|
|
help='The graph name')
|
|
parser.add_argument('--schema', required=True, type=str,
|
|
help='The schema of the graph')
|
|
parser.add_argument('--num-parts', required=True, type=int,
|
|
help='The number of partitions')
|
|
parser.add_argument('--output', required=True, type=str,
|
|
help='The output directory of the partitioned results')
|
|
parser.add_argument('--partitions-dir', help='directory of the partition-ids for each node type',
|
|
default=None, type=str)
|
|
|
|
#arguments needed for the distributed implementation
|
|
parser.add_argument('--world-size', help='no. of processes to spawn',
|
|
default=1, type=int, required=True)
|
|
params = parser.parse_args()
|
|
|
|
#invoke the pipeline function
|
|
logging.basicConfig(level='INFO', format=f"[{platform.node()} %(levelname)s %(asctime)s PID:%(process)d] %(message)s")
|
|
multi_machine_run(params)
|