Use pagerank_local_parallel rather than pagerank_local for better per…

…formance (#2945)

- use pagerank_local_parallel rather than pagerank_local

Signed-off-by: Tao He <linzhu.ht@alibaba-inc.com>

analytical_engine/test/run_vy_app.cc

@@ -35,7 +35,7 @@
 #include "lcc/lcc.h"
 #include "pagerank/pagerank.h"
 #include "pagerank/pagerank_auto.h"
-#include "pagerank/pagerank_local.h"
+#include "pagerank/pagerank_local_parallel.h"
 #include "sssp/sssp.h"
 #include "wcc/wcc.h"
 
@@ -321,7 +321,7 @@ void RunProjectedPR(std::shared_ptr<ProjectedFragmentType> fragment,
                     const std::string& out_prefix) {
   // using AppType = grape::PageRankAuto<ProjectedFragmentType>;
   using AppType = grape::PageRank<ProjectedFragmentType>;
-  // using AppType = grape::PageRankLocal<ProjectedFragmentType>;
+  // using AppType = grape::PageRankLocalParallel<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();

coordinator/gscoordinator/builtin/app/.gs_conf.yaml

@@ -1,8 +1,8 @@
 app:
   - algo: pagerank
     type: cpp_pie
-    class_name: grape::PageRankLocal
-    src: pagerank/pagerank_local.h
+    class_name: grape::PageRankLocalParallel
+    src: pagerank/pagerank_local_parallel.h
     compatible_graph:
       - grape::ImmutableEdgecutFragment
       - gs::ArrowProjectedFragment

python/graphscope/tests/conftest.py

@@ -693,6 +693,18 @@ def pagerank_local_result():
     yield ret
 
 
+@pytest.fixture(scope="module")
+def pagerank_local_parallel_result():
+    ret = {}
+    ret["directed"] = np.loadtxt(
+        "{}/ldbc/p2p-31-PR-LOCAL-PARALLEL-directed".format(property_dir), dtype=float
+    )
+    ret["undirected"] = np.loadtxt(
+        "{}/ldbc/p2p-31-PR-LOCAL-PARALLEL".format(property_dir), dtype=float
+    )
+    yield ret
+
+
 @pytest.fixture(scope="module")
 def hits_result():
     ret = {}

python/graphscope/tests/unittest/test_app.py

@@ -85,7 +85,7 @@ def test_run_app_on_pandas_graph(p2p_graph_from_pandas, sssp_result):
 def test_run_app_on_directed_graph(
     p2p_project_directed_graph,
     sssp_result,
-    pagerank_local_result,
+    pagerank_local_parallel_result,
     hits_result,
     bfs_result,
     clustering_result,
@@ -133,7 +133,7 @@ def test_run_app_on_directed_graph(
         .sort_values(by=["node"])
         .to_numpy(dtype=float)
     )
-    assert np.allclose(ret_pr, pagerank_local_result["directed"])
+    assert np.allclose(ret_pr, pagerank_local_parallel_result["directed"])
 
     # hits
     ctx_hits = hits(p2p_project_directed_graph, tolerance=0.001)
@@ -224,7 +224,7 @@ def test_run_app_on_directed_graph(
 def test_app_on_undirected_graph(
     p2p_project_undirected_graph,
     sssp_result,
-    pagerank_local_result,
+    pagerank_local_parallel_result,
     bfs_result,
     wcc_result,
     lpa_result,
@@ -262,14 +262,14 @@ def test_app_on_undirected_graph(
         .sort_values(by=["node"])
         .to_numpy(dtype=float)
     )
-    assert np.allclose(r2, pagerank_local_result["undirected"])
+    assert np.allclose(r2, pagerank_local_parallel_result["undirected"])
     ctx3 = pagerank(p2p_project_undirected_graph, 0.85, 10)
     r3 = (
         ctx3.to_dataframe({"node": "v.id", "r": "r"})
         .sort_values(by=["node"])
         .to_numpy(dtype=float)
     )
-    assert np.allclose(r3, pagerank_local_result["undirected"])
+    assert np.allclose(r3, pagerank_local_parallel_result["undirected"])
     # r4 = pagerank(arrow_project_graph, 10, 0.85) # check max_round=10
     # assert r4 is not None
     ctx5 = pagerank(p2p_project_undirected_graph, "0.85", "10")
@@ -278,29 +278,30 @@ def test_app_on_undirected_graph(
         .sort_values(by=["node"])
         .to_numpy(dtype=float)
     )
-    assert np.allclose(r5, pagerank_local_result["undirected"])
+    assert np.allclose(r5, pagerank_local_parallel_result["undirected"])
     ctx6 = pagerank(p2p_project_undirected_graph)
     r6 = (
         ctx6.to_dataframe({"node": "v.id", "r": "r"})
         .sort_values(by=["node"])
         .to_numpy(dtype=float)
     )
-    assert np.allclose(r6, pagerank_local_result["undirected"])
+    assert np.allclose(r6, pagerank_local_parallel_result["undirected"])
     assert np.allclose(
         ctx6.to_dataframe(
             {"node": "v.id", "r": "r"}, vertex_range={"begin": 1, "end": 4}
         )
         .sort_values(by=["node"])
         .to_numpy(),
         [
-            [1.0, 3.851369917786616],
-            [2.0, 5.808207281313435],
-            [3.0, 1.0284419953876565],
+            [1.0, 1.598779010550939e10],
+            [2.0, 2.117875105210911e10],
+            [3.0, 5.720363611374399e09],
         ],
     )
+
     assert np.allclose(
         sorted(ctx6.to_numpy("r", vertex_range={"begin": 1, "end": 4})),
-        sorted([3.851369917786616, 5.808207281313435, 1.0284419953876565]),
+        sorted([1.598779010550939e10, 2.117875105210911e10, 5.720363611374399e09]),
     )
 
     # bfs