par_to_file.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import networkx as nx\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from scipy import sparse\n",
    "from scipy.sparse import csr_matrix\n",
    "from scipy.sparse import coo_matrix\n",
    "from time import perf_counter\n",
    "import threading\n",
    "from multiprocessing import Process, Pool\n",
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "## use multi-thread method to implement graph partition, function defination\n",
    "def partion_multi_process (processID, graph_array, index_list, length):\n",
    "    t_m_p = perf_counter()\n",
    "    csr_row = np.int32(np.zeros(length + 1))\n",
    "    csr_col = []\n",
    "    temp_list = []\n",
    "    for i in range(len(graph_array)):\n",
    "        if ((int(graph_array[i][1]) >= index_list[processID]) and (int(graph_array[i][1]) < index_list[processID+1])):\n",
    "            ## csr\n",
    "            temp_list.append([graph_array[i][0], graph_array[i][1]])\n",
    "            csr_row[graph_array[i][0]+1] += 1\n",
    "    if (temp_list == []):\n",
    "        return [], []\n",
    "\n",
    "    temp_list = np.array(temp_list)\n",
    "    temp_list = temp_list[np.lexsort([temp_list.T[1]])] ## sort array by incremental order\n",
    "    temp_list = temp_list[np.lexsort([temp_list.T[0]])] ## sort array by incremental order\n",
    "\n",
    "    csr_row = np.add.accumulate(csr_row)\n",
    "    csr_col = temp_list[:, 1]\n",
    "\n",
    "    t_m_e = perf_counter() - t_m_p\n",
    "    print(\"partition finish, time : \", t_m_e)\n",
    "    \n",
    "    return csr_row, csr_col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Load data from hard-disk ... \n",
      "[[0 1]\n",
      " [0 2]\n",
      " [0 3]\n",
      " [0 4]\n",
      " [0 5]\n",
      " [1 2]\n",
      " [2 3]\n",
      " [2 4]\n",
      " [4 5]]\n"
     ]
    }
   ],
   "source": [
    "\n",
    "## filename = './datasets/as-skitter.txt'\n",
    "filename = './datasets/test.txt' ## seem our method performs better in amamzon dataset\n",
    "## filename = './datasets/roadNet-CA.txt'\n",
    "\n",
    "print(\"Load data from hard-disk ... \")\n",
    "txt_array_t = np.int64(np.loadtxt(filename))\n",
    "txt_array = txt_array_t[:,:2]\n",
    "print(txt_array)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Delete self-edges ... \n",
      "Delete duplicated edges ... \n",
      "Load data done \n"
     ]
    }
   ],
   "source": [
    "## if source_id > dest_id, exchange them; only suitable for undirected graph.\n",
    "for i in range (txt_array.shape[0]) :\n",
    "    if (txt_array[i][0] >= txt_array[i][1]) :\n",
    "        temp = txt_array[i][1]\n",
    "        txt_array[i][1] = txt_array[i][0]\n",
    "        txt_array[i][0] = temp\n",
    "\n",
    "txt_array = txt_array[np.lexsort([txt_array.T[1]])] ## sort array by incremental order\n",
    "txt_array = txt_array[np.lexsort([txt_array.T[0]])] ## sort array by incremental order\n",
    "\n",
    "print(\"Delete self-edges ... \")\n",
    "list_a = [] ## delete self-edges\n",
    "for i in range (txt_array.shape[0]) :\n",
    "    if (txt_array[i][0] == txt_array[i][1]) :\n",
    "        list_a.append(i)\n",
    "## print(list_a)\n",
    "## print(txt_array)\n",
    "txt_array = np.delete(txt_array, list_a, axis = 0)\n",
    "\n",
    "print(\"Delete duplicated edges ... \")\n",
    "list_b = []  ## delete duplicated edges \n",
    "for i in range (1, txt_array.shape[0]) :\n",
    "    if ((txt_array[i][0] == txt_array[i-1][0]) & (txt_array[i][1] == txt_array[i-1][1])) :\n",
    "        list_b.append(i)\n",
    "## print(list_a)\n",
    "txt_array = np.delete(txt_array, list_b, axis = 0)\n",
    "dest_array = txt_array[:, 1]\n",
    "dest_array = np.bincount(dest_array)\n",
    "dest_array = np.add.accumulate(dest_array)\n",
    "## print (dest_array.shape[0])\n",
    "# print (dest_array)\n",
    "\n",
    "print(\"Load data done \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "vertex range : 0 - 6 using process number : 2\n",
      "partition index array  [0 2 6]\n"
     ]
    }
   ],
   "source": [
    "## define the partition index, need to add partition function here\n",
    "t_partition_start = perf_counter()\n",
    "adj_matrix_dim = np.int64(txt_array.max()) + 1 ## get the max id for csr row size\n",
    "print (\"vertex range : 0 -\", adj_matrix_dim, end = \" \")\n",
    "partition_num = 2 ## can be set a variable, equals to thread numbers.\n",
    "print (\"using process number :\", partition_num)\n",
    "partition_index = np.zeros(partition_num + 1, dtype=np.int32)\n",
    "for i in range(partition_num - 1):\n",
    "    index_t = np.int32(len(txt_array)*(i+1)/partition_num)\n",
    "    abs_array = np.absolute(dest_array - index_t)\n",
    "    partition_index[i+1] = abs_array.argmin()\n",
    "    ## partition_index[i+1] = np.int32((adj_matrix_dim)*(i+1)/partition_num)\n",
    "    ## partition_index[i+1] = np.int32(adj_matrix_dim*(math.pow(0.717, partition_num-1-i)))\n",
    "partition_index[partition_num] = adj_matrix_dim\n",
    "print(\"partition index array \", partition_index)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "partition finish, time : partition finish, time :  0.001255318522453308 \n",
      "0.001018540933728218\n"
     ]
    }
   ],
   "source": [
    "## multi process pool.\n",
    "process_pool = Pool(partition_num)\n",
    "result_pool = []\n",
    "global _shared_array\n",
    "_shared_array = txt_array\n",
    "for i in range(partition_num):\n",
    "    result_pool.append(process_pool.apply_async(func=partion_multi_process, args=(i, _shared_array, partition_index, adj_matrix_dim)))\n",
    "\n",
    "process_pool.close()\n",
    "process_pool.join()\n",
    "\n",
    "graph_csr_row = []\n",
    "graph_csr_col = []\n",
    "\n",
    "for i in range(partition_num):\n",
    "    temp_a, temp_b = result_pool[i].get()\n",
    "    graph_csr_row.append(temp_a)\n",
    "    graph_csr_col.append(temp_b)\n",
    "    ## print(result_pool[i].get())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[array([0, 1, 1, 1, 1, 1, 1]), array([0, 4, 5, 7, 7, 8, 8])]\n",
      "[array([1]), array([2, 3, 4, 5, 2, 3, 4, 5])]\n"
     ]
    }
   ],
   "source": [
    "print(graph_csr_row)\n",
    "print(graph_csr_col)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2\n",
      "2\n"
     ]
    }
   ],
   "source": [
    "print(len(graph_csr_row))\n",
    "for i in range (0, len(graph_csr_row)):\n",
    "    fh = open(\"csr_row%s.txt\" % i, \"w\")\n",
    "    for j in range(0, len(graph_csr_row[i])):\n",
    "        fh.write(str(int(graph_csr_row[i][j])))\n",
    "        fh.write('\\n')\n",
    "    fh.close()\n",
    "\n",
    "print(len(graph_csr_col))\n",
    "for k in range (0, len(graph_csr_col)):\n",
    "    fk = open(\"csr_col%s.txt\" % k, \"w\")\n",
    "    for h in range(0, len(graph_csr_col[k])):\n",
    "        fk.write(str(int(graph_csr_col[k][h])))\n",
    "        fk.write('\\n')\n",
    "    fk.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.15"
  },
  "vscode": {
   "interpreter": {
    "hash": "d8c35ca727124e6ef49ad1437a456b0984b9ec489fab4181e991d3ddd54fc8fa"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}