kylasa
4b87e47f15
* Initial implementation of Distributed data processing step in the Distributed Training pipeline Implemented the following: 1) Read the output of parmetis (node-id to partition-id mappings) 2) Read the original graph files 3) Shuffle the node/edge metadata and features 4) output the partition specific files in DGL format using convert_partition.py functionality 5) Graph meta data is serialized in json format on rank-0 machine. * Bug Fixes identified during verification of the dataset 1. When sending out global-id lookups for non-local nodes, in the msg_alltoall.py, conditional filter was used to identify the indices in node_data which is incorrect. Replaced the conditional filter with intersect1d to find out the common node ids and appropriate indices which are later used to identify the needed information to communicate. 2. When writing the graph level json file in distributed processing, the edge_offset on non-rank-0 machines was starting from 0 instead of the appropriate offset. Now added code to start the edge(s) from correct starting offset instead of 0 always. * Restructuring and consolidation of code 1) Fixed issue when running verify_mag_dataset.py, Now we read xxx_removed_edges.txt and add these edges to `edge_data`. This will ensure that the self-loops and duplicate edges are handling appropriately when compared to the original dataset. 2) Consolidated code into a fewer files and changed code to following the python naming convention. * Code changes addressing code review comments Following changes are made in this commit. 1) Naming convention is defined and code is changed accordingly. Definition of various global_ids are defined and how to read them is mentioned. 2) All the code review comments are addressed 3)Files are moved to a new directory with dgl/tools directory as per suggestion 4) README.md file is include and it contains detailed information about the Naming convention adopted by the code, high level overview of the algorithm used in data-shuffling, example command-line to use on a single machine. * addressing github review comments Made code changes addressing all the review comments from GitHub. * Addressing latest code review comments Addressed all the latest code reviewing comments. One of the major changes is treating the node and edge metadata as dictionary objects and removing all the python lists with numpy arrays. * Update README.md Text rendering corrections * Addressed code review comments Addressed code review comments for the latest code review Co-authored-by: xiang song(charlie.song) <classicxsong@gmail.com>
4.9 KiB
xxx_nodes.txt format
This file is used to provide node information to this framework. Following is the format for each line in this file:
<node_type> <weight1> <weight2> <weight3> <weight4> <global_type_node_id> <attributes>
where node_type is the type id of this node, weights can be any number of columns as determined by the user, global_type_node_id are the contiguous ids starting from 0 for a particular node_type. And attributes can be any number of columns at the end of each line.
###xxx___edges.txt format This file is used to provide edge information to this framework. Following is the format for each line in this file:
<global_src_id> <global_dst_id> <global_type_edge_id> <edge_type> <attributes>
where global_src_id and global_dst_id are two end points of an edge, global_type_edge_id is the unique id assigned to each edge type and are contiguous, and starting from 0, for each edge_type. Attributes can be any number of columns at the end of each line.
###Naming convention
global_ prefix (for any node or edge ids) indicate that these ids are read from graph input files. These ids are allocated to nodes and edges before data shuffling. These ids are globally unique across all partitions.
shuffle_global_ prefix (for any node or edge ids) indicate that these ids are assigned after the data shuffling is completed. These ids are globally unique across all partitions.
part_local_ prefix (for any node or edge ids) indicate that these ids are assigned after the data shuffling and are unique within a given partition.
For instance, if a variable is named as global_src_id it means that this id is read from the graph input file and is assumed to be globally unique across all partitions. Similarly if a variable is named part_local_node_id then it means that this node_id is assigned after the data shuffling is complete and is unique with a given partition.
###High level description of the algorithm
####Single file format for graph input files
Here we assume that all the nodes' related data is present in one single file and similarly all the edges are in one single file.
In this case following steps are executed to write dgl objects for each partition, as assigned my any partitioning algorithm, for example METIS.
#####Step 1 (Data Loading):
Rank-0 process reads in all the graph files which are xxx_nodes.txt, xxx_edges.txt, node_feats.dgl, edge_feats.dgl and xxx_removed_edges.txt.
Rank-0 process determines the ownership of nodes by using the output of partitioning algorithm (here, we expect the output of partitioning step is a mapping between a node and its partition id for the entire graph). Edge ownership is determined by the destination node-id for that edge. Each edge belongs to the partition-id of the destination node-id of each edge.
#####Step 2 (Data Shuffling):
Rank-0 process will send node-data, edge-data, node-features, edge-features to their respective processes by using the ownership rules described in Step-1. Non-Rank-0 processes will receive their own nodes, edges, node-features and edge-features and store them in local data-structures. Upon completion of sending information Rank-0 process will delete nodes, edges, node-features and edge-features which are not owned by rank-0.
#####Step 3 (ID assignment and resolution):
At this time all the ranks will have their own local information in their respective data structures. Then each process will perform the following steps: a) Assign shuffle_global_xxx (here xxx is node_ids and edge_ids) for nodes and edges by performing prefix sum on all ranks. b) Assign part_local_xxx (xxx means node_ids and edge_ids) to nodes and edges so that they can be used to index into the node and edge features, and c) Retrieve shuffle_global_node_ids by using global_node_ids to determine the ownership of any given node. This step is done for the node_ids (present locally on any given rank) for which shuffle_global_node_ids were assigned on a different rank'ed process.
#####Ste 4 (Serialization):
After every rank has global-ids, shuffle_global-ids, part_local-ids for all the nodes and edges present locally, then it proceeds by DGL object creation. Finally Rank-0 process will aggregate graph-level metadata and create a json file with graph-level information.
###How to use this tool To run this code on a single machine using multiple processes, use the following command
python3 data_proc_pipeline.py --world-size 2 --nodes-file mag_nodes.txt --edges-file mag_edges.txt --node-feats-file node_feat.dgl --metis-partitions mag_part.2 --input-dir /home/ubuntu/data --graph-name mag --schema mag.json --num-parts 2 --num-node-weights 4 --workspace /home/ubuntu/data --node-attr-dtype float --output /home/ubuntu/data/outputs --removed-edges mag_removed_edges.txt
Above command, assumes that there are 2 partitions and number of node weights are 4. All other command line arguments are self-explanatory.