* Replacing numpy's unique with custom implementation
* Added docstring to the new function.
* Adding unit tests
* Numpy's version issues with the 'kind' argument.
* Addressing CI Test Failure.
* Addressing CI review comments.
* revised implementation, optimized for time.
* added missing arguments for fallback case.
* Addressing CI test failures.
* Resolving issues with PYTHONPATH
* Fix CI Test Failure issues.
* fix CI test failures.
---------
Co-authored-by: Rhett Ying <85214957+Rhett-Ying@users.noreply.github.com>
* Sync parmetis_wrapper with changes in metadata.json
1. In the preprocess.py, make sure that num_partitions is defined as input argument. Also, align 'input_dir' with the input dataset. schema_file is assumed to be located inside the input_dir. Also, graph_stats.txt file is assumed to be present in the input_dir.
2. Use DGL_HOME environment variable so that parmetis_wrapper command can be run anywhere.
* Fix CI test failure cases.
* Addressing CI review comments.
* Addressing CI test failures.
* Applying lintrunner patch
* Support for no. of chunks smaller than no. of partitions and Adding appropriate test cases.
Following changes are made with this PR.
1. Code changes for handling no. of chunks smaller than no. of partitions
2. Adding new test cases, which were previously deleted, for no. of chunks smaller than no. of partitions.
3. Also adding test cases, where multiple partitions are handled by a single process.
* Committing the missing files in this commit.
* lintrunner patch.
* lintrunner check
* lintrunner patch here.
* CI review comments.
Handling corner cases in the distributed lookup service. When the get partition ids function is invoked with empty request. This is needed because we are using alltoall function in the get_partition_ids function.
* Implemented the following changes.
* Remove NUM_NODES_PER_CHUNK
* Remove NUM_EDGES_PER_CHUNK
* Remove the dependency between no. of edge files per edge type and no. of partitions
* Remove the dependency between no. of edge feature files per edge type and no. of partitions
* Remove the dependency between no. of edge feature files and no. of edge files per edge type.
* Remove the dependency between no. of node feature files and no. of partitions
* Add “node_type_counts”. This will be a list of integers. Each integer will represent total count of a node-type. The index in this list and the index in the “node_type” will be the same for a given node-type.
* Add “edge_type_counts”. This will be a list of integers. Each integer will represent total count of an edge-type. The index in this list and the index in the “edge_type” list will be the same for a given edge-type.
* Applying lintrunner patch.
* Adding missing keys to the metadata in the unit test framework.
* lintrunner patch.
* Resolving CI test failures due to merge conflicts.
* Applying lintrunner patch
* applying lintrunner patch
* Replacing tabspace with spaces - to satisfy lintrunner
* Fixing the CI Test Failure cases.
* Applying lintrunner patch
* lintrunner complaining about a blank line.
* Resolving issues with print statement for NoneType
* Removed tests for the arbitrary chunks tests. Since this functionality is not supported anymore.
* Addressing CI review comments.
* addressing CI review comments
* lintrunner patch
* lintrunner patch.
* Addressing CI review comments.
* lintrunner patch.
* Uneven distribution of nodes/edges/features
To handle unevenly sized files for nodes/edges and feature files for nodes and edges, we have to synchronize before starting large no. of messages (either one large message or a burst of messages).
* Applying lintrunner patch.
* Removing tabspaces for lintrunner.
* lintrunner patch.
* removed issues introduced by the merge conflicts. Lots of code was repeated
* get_partition_ids, service provided by the distributed lookup service has high memory footprint
'get_partitionid' function, which is used to retrieve owner processes of the given list of global node ids, has high memory footprint. Currently this is of the order of 8x compared to the size of the input list.
For massively large datasets, this memory needs are very unrealistic and may result in OOM. In the case of CoreGraph, when retrieving owner of an edge list of size 6 Billion edges, the memory needs can be as high as 8*8*8 = 256 GB.
To limit the amount of memory used by this function, we split the size of the message sent to the distributed lookup service, so that each message is limited by the number of global node ids, which is 200 million. This reduced the memory footprint of this entire function to be no more than 0.2 * 8 * 8 = 13 GB. which is within reasonable limits.
Now since we send multiple small messages compared to one large message to the distributed lookup service, this may consume more wall-clock-time compared to earlier implementation.
* lintrunner patch.
* using np.ceil() per suggestion.
* converting the output of np.ceil() as ints.
* Wrap np.argsort() in a function. This
Use a python wrapper for the np.argsort() function for better usage of systems memory.
* lintrunner patch.
* lintrunner patch.
* Changes to address code review comments.
* Rehash code to optimize for loop
Reduced number of instructions in for loop, which exchanging edge features. This will reduce the number of times numpy's intersect1d is invoked (saving the runtime and memory overhead needs of numpy).
* Applying lintrunner patch to data_shuffle.py
* Edge Ownership processes are computed on the fly when required.
Earlier we were storing Edge ownership processes after the dataset was retrieved from the disk. For massively large datasets, each node can handle upto 5 Billion edges, this means storing owner process-ids will consume 5 * 8 = 40GB. This memory will be hanging around until the edges are exchanged.
To reduce the memory footprint of the pipeline, we no longer store the ownership process-ids in the 'edge_data' dictionary after reading the dataset from the disk. Instead, we compute them on the fly at the time of exchanging edges.
Another optimization is not to send/receive all the messages in a one single large message. Instead we now split the total number edges into chunks, limited by 8 GB per node. And we iterate until all the chunks are exchanged.
Once all the edges are exchanged, as a sanity check, we compute the total number of edges in the system and compare it with the original value before edge shuffling, in a final assert statement before return the result to the caller.
* Applying lintrunner patch.
Following changes are made in this PR.
1. In dataset_utils.py, when reading edges from disk we follow the order defined by the STR_EDGE_TYPE key in the metadata.json file. This order is implicitly used to assign edgeid to edge types. This same order is used to read edges from the disk as well.
2. Now the unit test framework will also randomize the order of edges read from the disk. This is done for the edges when reading from the disk for the unit tests.
Co-authored-by: Quan (Andy) Gan <coin2028@hotmail.com>
* Allow reading and writing single-column vector Parquet files.
These files are commonly produced by Spark ML's feature processing code.
* [Dist] Only write single-column vector files for Parquet in tests.
* [Dist] Add support for Parquet-formatted edges files, remove some assumptions on edge file number.
* [Dist] Add parquet edges option to unit tests.
Co-authored-by: xiang song(charlie.song) <classicxsong@gmail.com>
* [Dist] enable to chunk node/edge data into arbitrary number of chunks
* [Dist] enable to split node/edge data into arbitrary parts
* refine code
* Format boolean to uint8 forcely to avoid dist.scatter failure
* convert boolean to int8 before scatter and revert it after scatter
* refine code
* fix test
* refine code
* move test utilities into utils.py
* update comment
* fix empty data
* update
* update
* fix empty data issue
* release unnecessary mem
* release unnecessary mem
* release unnecessary mem
* release unnecessary mem
* release unnecessary mem
* remove unnecessary shuffle data
* separate array_split into standalone utility
* add example
Co-authored-by: xiang song(charlie.song) <classicxsong@gmail.com>
* Flexible pipeline - Initial commit
1. Implementation of flexible pipeline feature.
2. With this implementation, the pipeline now supports multiple partitions per process. And also assumes that num_partitions is always a multiple of num_processes.
* Update test_dist_part.py
* Code changes to address review comments
* Code refactoring of exchange_features function into two functions for better readability
* Upadting test_dist_part to fix merge issues with the master branch
* corrected variable names...
* Fixed code refactoring issues.
* Provide missing function arguments to exchange_feature function
* Providing the missing function argument to fix error.
* Provide missing function argument to 'get_shuffle_nids' function.
* Repositioned a variable within its scope.
* Removed tab space which is causing the indentation problem
* Fix issue with the CI test framework, which is the root cause for the failure of the CI tests.
1. Now we read files specific to the partition-id and store this data separately, identified by the local_part_id, in the local process.
2. Similarly as above, we also differentiate the node and edge features type_ids with the same keys as above.
3. These above two changes will help up to get the appropriate feature data during the feature exchange and send to the destination process correctly.
* Correct the parametrization for the CI unit test cases.
* Addressing Rui's code review comments.
* Addressing code review comments.
* Send out the message to the distributed lookup service in batches.
* Update function signature for allgather_sizes function call.
* Removed the unnecessary if statement .
* Removed logging.info message, which is not needed.
* [Dist] deprecate etype and always use canonical etype for partition and load
* enable canonical etypes in dist part pipeline
* resolve rebase conflicts
* fix lint
* fix test failure
* throw exception if outdated part config is loaded
* refine
* refine
* revert unnecessary change
* fix typo
* [Dist] Reduce peak memory in DistDGL: avoid validation, release memory once loaded
* remove orig_id from ndata/edata for partition_graph()
* delete orig_id from ndata/edata in dist part pipeline
* reduce dtype size and format before saving graphs
* fix lint
* ETYPE requires to be int32/64 for CSRSortByTag
* fix test failure
* refine
* Creating ParMETIS wrapper script to run parmetis using one script from user perspective
* Addressed all the CI comments from PR https://github.com/dmlc/dgl/pull/4529
* Addressing CI comments.
* Isort, and black changes.
* Replaced python with python3
* Replaced single quote with double quotes per suggestion.
* Removed print statement
* Addressing CI Commets.
* Addressing CI review comments.
* Addressing CI comments as per chime discussion with Rui
* CI Comments, Black and isort changes
* Align with code refactoring, black, isort and code review comments.
* Addressing CI review comments, and fixing merge issues with the master branch
* Updated with proper unit test skip decorator
* Added support for edge features.
* Added comments and removing unnecessary print statements.
* updated data_shuffle.py to remove compile error.
* Repaled python3 with python to match CI test framework.
* Removed unrelated files from the pull request.
* Isort changes.
* black changes on this file.
* Addressing CI review comments.
* Addressing CI comments.
* Removed duplicated and resolved merge conflict code.
* Addressing CI Comments from Rui.
* Addressing CI comments, and fixing merge issues.
* Addressing CI comments, code refactoring, isort and black
* [Dist] enable to partition many chunks into less partitions via pipeline
* refine
* add meta file for num_parts, add more tests, refine docstring
* remove args.num_parts
* create pydantic class for partition metadata
* refine
* rename json file
* [DistPart] expose timeout config for process group
* refine code
* Update tools/distpartitioning/data_proc_pipeline.py
Co-authored-by: Minjie Wang <wmjlyjemaine@gmail.com>
Co-authored-by: Minjie Wang <wmjlyjemaine@gmail.com>
* Distributed Lookup service which is for retrieving global_nids to shuffle-global-nids/partition-id mappings
1. Implemented a class to provide distributed lookup service
2. This class can be used to retrieve global-nids mappings
* Code changes to address CI comments.
1. Removed some unneeded type_casts to numpy.int64
2. Added additional comments when iterating over the partition-ids list.
3.Added docstring to the class and adjusted comments where it is relevant.
* Updated code comments and variable names...
1. Changed the variable names to appropriately represent the values stored in these variables.
2. Updated the docstring correctly.
* Corrected docstring as per the suggestion... and removed all the capital letters for Global nids and Shuffle Global nids...
* Addressing CI review comments.
* 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>
* Fix for node_subgraph function, which seems to generate segmentation fault for very large partitions
1. Removed three graph dgl objects and we create the final dgl object directly by maintaining the following constraints
a) nodes are reordered so that local nodes are placed in the beginning of the nodes list compared to non-local nodes.
b)Edges order are maintained as passed into this function.
c) src/dst end points are mapped to target values based on the reshuffle'd nodes order.
* Code changes addressing CI comments for this PR
1. Used Da's suggested map to map nodes from old to new order.
This is much simpler and mem. efficient.
* Addressing CI Comments.
1. Reduced the amount of documentation to reflect the actual implementation.
2. named the mapping object appropriately.
* Alltoall Fix to bypass gloo - alltoallv bug which is preventing further testing
1. Replaced alltoallv gloo wrapper call with alltoall message.
2. All the messages are padded to be of same length
3. Receiving side unpads the messages and continues processing.
* Code changes to address CI comments
1. Removed unused functions from gloo_wrapper.py
2. Changed the function signature of alltoallv_cpu_data as suggested.
3. Added docstring to include more description of the functionality inside alltoallv_cpu_data. Included more asserts to validate the assumptions.
* Changed the function name appropriately
Changed the function name from "alltoallv_cpu_data" to alltoallv_cpu which I believe is appropriate because underlying functionality is providing alltoallv which is basically alltoall_cpu + padding
* Added code and text to address the review comments.
1. Changed the function name to indicate the local use of this function.
2. Changed docstring to indicate the assumptions made by alltoallv_cpu function.
* Removed unused function from import statement
Removed unused/removed function from import statement.
* Code changes to address the updated file format support for massively large graphs.
1. Updated the docstring for the starting function 'gen_dist_partitions" to describe the newly proposed file format for input dataset.
2. Code which was dependent on the structure of the old-metadata json object has been updated to read from the newly proposed metadata file.
3. Fixed some errors when appropriate functions were invoked and the calling function expects return values from the invoked furnction.
4. This modified code has been tested on "mag" dataset using 4-way partitions and verified the results
* Code changes to address the CI review comments
1. Improved docstrings for some functions.
2. Added a new function in the utils.py to compute the id ranges and this is used in multiple places.
* Added TODO to indicate the redundant data structure.
Because of the new file format changes, one of the dictionaries (node_feature_tids, node_tids) will be redundant. Added TODO text so that this will be removed in the next iteration of code changes.
* Code changes for the following
1. Generating node data at each process
2. Reading csv files using pyarrow
3. feature complete code.
* Removed some typo's because of which unit tests were failing
1. Change the file name to correct file name when loading edges from file
2. When storing node-features after shuffling, use the correct key to store the global-nids of node features which are received after transmitted.
* Code changes to address CI comments by reviewers
1. Removed some redundant code and added text in the doc-strings to describe the functionality of some functions.
2 function signatures and invocations now match w.r.t argument list
3. Added detailed description of the metadata json structure so that the users understand the the type of information present in this file and how it is used through out the code.
* Addressing code review comments
1. Addressed all the CI comments and some of the changes include simplifying the code related to the concatenation of lists and enhancing the docstrings of functions which are changed in this process.
* Update docstring's of two functions appropriately in response to code review comments
Removed "todo" from the docstring of the gen_nodedata function.
Added "todo" to the gen_dist_partitions function when node-id to partition-id's are read for the first time.
Removed 'num-node-weights' from the docstring for the get_dataset function and added schema_map docstring to the argument list.
* Added code to support multiple-file-support feature and removed single-file-support code
1. Added code to read dataset in multiple-file-format
2. Removed code for single-file format
* added files missing in the previous commit
This commit includes dataset_utils.py, which reads the dataset in multiple-file-format, gloo_wrapper function calls to support exchanging dictionaries as objects and helper functions in utils.py
* Update convert_partition.py
Updated function call "create_metadata_json" file to include partition_id so that each rank only creates its own metadata object and later on these are accumulated on rank-0 to create graph-level metadata json file.
* addressing code review comments during the CI process
code changes resulting from the code review comments received during the CI process.
* Code reorganization
Addressing CI comments and code reorganization for easier understanding.
* Removed commented out line
removed commented out line.
* code changes for bug fixes identified during mag_lsc dataset
1. Changed from call torch.Tensor() to torch.from_numpy() to address memory corruption issues when creating large tensors. Tricky thing is this works correctly for small tensors.
2. Changed dgl.graph() function call to include 'num_nodes" argument to specifically mention all the nodes in a graph partition.
* Update convert_partition.py
Moving the changes to the function "create_metadata_json" function to the "multiple-file-format" support, where this change is more appropriate. Since multiple machine testing was done with these code changes.
* Addressing review comments.
Removed space as suggested at the end of the line
kylasa