Format

Committed-by: acezen from Dev container

Author: acezen

python/graphscope/__init__.py

@@ -49,7 +49,6 @@
 from graphscope.framework.errors import *
 from graphscope.framework.graph import Graph
 from graphscope.framework.graph_builder import load_from
-from graphscope.framework.graph_builder import load_from_gar
 from graphscope.version import __version__
 
 __doc__ = """

python/graphscope/client/session.py

@@ -1293,13 +1293,6 @@ def load_from(self, *args, **kwargs):
         with default_session(self):
             return graphscope.load_from(*args, **kwargs)
 
-    def load_from_gar(self, *args, **kwargs):
-        """Load a graph from gar format files within the session.
-        See more information in :meth:`graphscope.load_from_gar`.
-        """
-        with default_session(self):
-            return graphscope.load_from_gar(*args, **kwargs)
-
     def _run_on_local(self):
         self._config_params["port"] = None
         self._config_params["vineyard_socket"] = ""

python/graphscope/framework/dag_utils.py

@@ -1077,8 +1077,8 @@ def fetch_gremlin_result(result_set, fetch_type="one"):
     return op
 
 
-def archive_graph(graph, path):
-    """Archive a graph to gar format with a path.
+def save_to_graphar(graph, path):
+    """Save a graph to graphar format with a path.
 
     Args:
         graph (:class:`graphscope.framework.graph.GraphDAGNode`): Source graph.
@@ -1105,11 +1105,7 @@ def archive_graph(graph, path):
     return op
 
 
-def serialize_graph(
-    graph,
-    path: str,
-    **kwargs
-):
+def serialize_graph(graph, path: str, **kwargs):
     """Serialize graph to the specified location
        The meta and data of graph is dumped to specified location,
        and can be restored by `Graph.load_from` in other sessions.

python/graphscope/framework/graph.py

@@ -103,9 +103,6 @@ def save_to(self, path, **kwargs):
     def load_from(cls, path, sess, **kwargs):
         raise NotImplementedError
 
-    def archive(self, path, **kwargs):
-        raise NotImplementedError
-
     @abstractmethod
     def project(self, vertices, edges):
         raise NotImplementedError
@@ -428,16 +425,6 @@ def to_dataframe(self, selector, vertex_range=None):
         op = dag_utils.graph_to_dataframe(self, selector, vertex_range)
         return ResultDAGNode(self, op)
 
-    def archive(self, path):
-        """Archive the graph to gar format with graph yaml file path.
-
-        Args:
-            path (str): The graph yaml file path describe how to archive the graph.
-        """
-        check_argument(self.graph_type == graph_def_pb2.ARROW_PROPERTY)
-        op = dag_utils.archive_graph(self, path)
-        return ArchivedGraph(self._session, op)
-
     def to_directed(self):
         op = dag_utils.to_directed(self)
         graph_dag_node = GraphDAGNode(self._session, op)
@@ -1104,6 +1091,7 @@ def load_from(cls, uristring, sess=None):
             `Graph`: A new graph object. Schema and data is supposed to be
         """
         from graphscope.client.session import get_default_session
+
         if sess is None:
             sess = get_default_session()
         uri = urlparse(uristring)
@@ -1116,19 +1104,20 @@ def load_from(cls, uristring, sess=None):
                 raise ValueError("Unknown source: %s" % source)
         else:
             # not a uri string, assume it is a path for deserialization
-            op = dag_utils.deserialize_graph(path, sess, **kwargs)
+            op = dag_utils.deserialize_graph(path, sess)
             return sess._wrapper(GraphDAGNode(sess, op))
 
-    def save_to(self,
-                path,
-                format="serialization",
-                graphar_graph_name="graph",
-                graphar_vertex_block_size=262144,  # 2^18
-                graphar_edge_block_size=4194304,   # 2^22
-                graphar_file_format="parquet",
-                graphar_version="v1",
-                ):
-        """ Save graph to specified location with specified format.
+    def save_to(
+        self,
+        path,
+        format="serialization",
+        graphar_graph_name="graph",
+        graphar_vertex_block_size=262144,  # 2^18
+        graphar_edge_block_size=4194304,  # 2^22
+        graphar_file_format="parquet",
+        graphar_version="v1",
+    ):
+        """Save graph to specified location with specified format.
 
         Args:
             path (str): the directory path to write graph.
@@ -1143,12 +1132,21 @@ def save_to(self,
         """
         if format == "graphar":
             graph_info_path = utils.generate_graphar_info_from_schema(
-                path, self._schema, graphar_graph_name, graphar_vertex_block_size, graphar_edge_block_size, graphar_file_format, graphar_version)
-            self._session._wrapper(self._graph_node.archive(graph_info_path))
+                path,
+                self._schema,
+                graphar_graph_name,
+                graphar_vertex_block_size,
+                graphar_edge_block_size,
+                graphar_file_format,
+                graphar_version,
+            )
+            op = dag_utils.save_to_graphar(self, graph_info_path)
+            self._session.dag.add_op(op)
+            self._session._wrapper(op)
             return {"type": format, "uri": "graphar+" + graph_info_path}
         elif format == "serialization":
             # serialize graph
-            op = dag_utils.serialize_graph(self, path, **kwargs)
+            op = dag_utils.serialize_graph(self, path)
             self._session.dag.add_op(op)
             self._session._wrapper(op)
             return {"type": format, "uri": path}
@@ -1228,13 +1226,3 @@ def __init__(self, session, op):
         self._op = op
         # add op to dag
         self._session.dag.add_op(self._op)
-
-
-class ArchivedGraph(DAGNode):
-    """Archived graph node in a DAG"""
-
-    def __init__(self, session, op):
-        self._session = session
-        self._op = op
-        # add op to dag
-        self._session.dag.add_op(self._op)

python/graphscope/framework/graph_builder.py

@@ -205,46 +205,3 @@ def load_from(
         compact_edges=compact_edges,
     )
     return graph
-
-
-def load_from_gar(
-    graph_info_path: str,
-    directed=True,
-    oid_type="int64_t",
-    vertex_map="global",
-    compact_edges=False,
-) -> Graph:
-    sess = get_default_session()
-    oid_type = utils.normalize_data_type_str(oid_type)
-    if oid_type not in ("int32_t", "int64_t", "std::string"):
-        raise ValueError("The 'oid_type' can only be int32_t, int64_t or string.")
-    if compact_edges:
-        raise ValueError(
-            "Loading from gar with 'compact_edges' hasn't been supported yet."
-        )
-    # generate and add a loader op to dag
-    vertex_map = utils.vertex_map_type_to_enum(vertex_map)
-    # construct create graph op
-    config = {
-        types_pb2.DIRECTED: utils.b_to_attr(directed),
-        types_pb2.OID_TYPE: utils.s_to_attr(oid_type),
-        types_pb2.GENERATE_EID: utils.b_to_attr(False),
-        types_pb2.RETAIN_OID: utils.b_to_attr(False),
-        types_pb2.VID_TYPE: utils.s_to_attr("uint64_t"),
-        types_pb2.IS_FROM_VINEYARD_ID: utils.b_to_attr(False),
-        types_pb2.IS_FROM_GAR: utils.b_to_attr(True),
-        types_pb2.VERTEX_MAP_TYPE: utils.i_to_attr(vertex_map),
-        types_pb2.COMPACT_EDGES: utils.b_to_attr(compact_edges),
-        types_pb2.GRAPH_INFO_PATH: utils.s_to_attr(graph_info_path),
-    }
-    op = dag_utils.create_graph(
-        sess.session_id, graph_def_pb2.ARROW_PROPERTY, inputs=[], attrs=config
-    )
-    graph = sess.g(
-        op,
-        oid_type=oid_type,
-        directed=directed,
-        vertex_map=vertex_map,
-        compact_edges=compact_edges,
-    )
-    return graph

python/graphscope/framework/utils.py

@@ -25,7 +25,6 @@
 import socket
 import string
 import subprocess
-import yaml
 import tempfile
 import threading
 import time
@@ -36,6 +35,7 @@
 import numpy as np
 import pandas as pd
 import psutil
+import yaml
 from google.protobuf.any_pb2 import Any
 
 from graphscope.client.archive import OutArchive
@@ -685,15 +685,10 @@ def _to_numpy_dtype(dtype):
 
 
 def generate_graphar_info_from_schema(
-        path,
-        schema,
-        graph_name,
-        vertex_block_size,
-        edge_block_size,
-        file_type,
-        version):
-
+    path, schema, graph_name, vertex_block_size, edge_block_size, file_type, version
+):
     import copy
+
     class Dumper(yaml.Dumper):
         def increase_indent(self, flow=False, indentless=False):
             return super(Dumper, self).increase_indent(flow, False)
@@ -726,31 +721,53 @@ def PbDataType2InfoType(str):
         info["version"] = "gar/{}".format(version)
         info["property_groups"] = [{"properties": [], "file_type": file_type}]
         for property in schema.get_vertex_properties(vertex_label):
-            info["property_groups"][0]["properties"].append({"name": property.name, "data_type": PbDataType2InfoType(graph_def_pb2.DataTypePb.Name(property.data_type)), "is_primary": True if property.name == "id" else False})
-        output_path = os.path.join(path, vertex_label+".vertex.yml")
+            info["property_groups"][0]["properties"].append(
+                {
+                    "name": property.name,
+                    "data_type": PbDataType2InfoType(
+                        graph_def_pb2.DataTypePb.Name(property.data_type)
+                    ),
+                    "is_primary": True if property.name == "id" else False,
+                }
+            )
+        output_path = os.path.join(path, vertex_label + ".vertex.yml")
         with open(output_path, "w") as f:
             yaml.dump(info, f, Dumper=Dumper, default_flow_style=False)
-        graph_info["vertices"].append(vertex_label+".vertex.yml")
+        graph_info["vertices"].append(vertex_label + ".vertex.yml")
     # process edge info
     for edge_label in schema.edge_labels:
         properties = []
         for property in schema.get_edge_properties(edge_label):
-            properties.append({"name": property.name, "data_type": PbDataType2InfoType(graph_def_pb2.DataTypePb.Name(property.data_type)), "is_primary": False})
+            properties.append(
+                {
+                    "name": property.name,
+                    "data_type": PbDataType2InfoType(
+                        graph_def_pb2.DataTypePb.Name(property.data_type)
+                    ),
+                    "is_primary": False,
+                }
+            )
         csr_adj_list = {
             "file_type": file_type,
-            "property_groups": [{"properties": copy.deepcopy(properties), "file_type": file_type}],
+            "property_groups": [
+                {"properties": copy.deepcopy(properties), "file_type": file_type}
+            ],
             "ordered": True,
             "aligned_by": "src",
         }
         csc_adj_list = {
             "file_type": file_type,
-            "property_groups": [{"properties": copy.deepcopy(properties), "file_type": file_type}],
+            "property_groups": [
+                {"properties": copy.deepcopy(properties), "file_type": file_type}
+            ],
             "ordered": True,
             "aligned_by": "dst",
         }
         for r in schema.get_relationships(edge_label):
             info = dict()
-            info["prefix"] = "edge/" + r.source + "_" + edge_label + "_" + r.destination + "/"
+            info["prefix"] = (
+                "edge/" + r.source + "_" + edge_label + "_" + r.destination + "/"
+            )
             info["edge_label"] = edge_label
             info["src_label"] = r.source
             info["dst_label"] = r.destination
@@ -759,10 +776,14 @@ def PbDataType2InfoType(str):
             info["dst_chunk_size"] = vertex_block_size
             info["version"] = "gar/{}".format(version)
             info["adj_lists"] = [csr_adj_list, csc_adj_list]
-            output_path = os.path.join(path, r.source + "_" + edge_label + "_" + r.destination + ".edge.yml")
+            output_path = os.path.join(
+                path, r.source + "_" + edge_label + "_" + r.destination + ".edge.yml"
+            )
             with open(output_path, "w") as f:
                 yaml.dump(info, f, Dumper=Dumper, default_flow_style=False)
-        graph_info["edges"].append(r.source + "_" + edge_label + "_" + r.destination + ".edge.yml")
+        graph_info["edges"].append(
+            r.source + "_" + edge_label + "_" + r.destination + ".edge.yml"
+        )
     graph_info_path = os.path.join(path, graph_name + ".graph.yml")
     with open(graph_info_path, "w") as f:
         yaml.dump(graph_info, f, Dumper=Dumper, default_flow_style=False)

python/graphscope/tests/unittest/test_gar.py

@@ -18,19 +18,21 @@
 
 import os
 
+from graphscope.framework.graph import Graph
+
 gar_test_repo_dir = os.path.expandvars("${GS_TEST_DIR}")
 
 
 def test_load_from_gar(graphscope_session):
     graph_yaml = os.path.join(
         gar_test_repo_dir, "graphar/ldbc_sample/parquet/ldbc_sample.graph.yml"
     )
-    print(graph_yaml)
-    graph = graphscope_session.load_from_gar(graph_yaml)
+    graph_yaml_path = "graphar+file://" + graph_yaml
+    print(graph_yaml_path)
+    g = Graph.load_from(graph_yaml_path)
     assert graph.schema is not None
     del graph
 
 
-def test_archive_to_gar(ldbc_graph):
-    graph_yaml = os.path.join(gar_test_repo_dir, "graphar/ldbc/ldbc.graph.yml")
-    ldbc_graph.archive(graph_yaml)
+def test_save_to_graphar(ldbc_graph):
+    ldbc_graph.save_to("/tmp/")