Method output

python/graphscope/framework/context.py:232–276

Dump results to `fd`. Support dumps data to local (respect to pod) files, hdfs or oss. It first write results to a vineyard dataframe, and let vineyard do the data dumping job. `fd` must meet specific formats, with auth information if needed. As follows:

(self, fd, selector, vertex_range=None, **kwargs)

        return ResultDAGNode(self, op)

    def output(self, fd, selector, vertex_range=None, **kwargs):
        """Dump results to `fd`.
        Support dumps data to local (respect to pod) files, hdfs or oss.
        It first write results to a vineyard dataframe, and let vineyard
        do the data dumping job.
        `fd` must meet specific formats, with auth information if needed. As follows:

            - local
                `file:///tmp/result_path`
            - oss
                `oss:///bucket/object`
            - hdfs
                `hdfs:///tmp/result_path`

        Args:
            fd (str): Output location.
            selector (dict): Similar to `to_dataframe`.
            vertex_range (dict, optional): Similar to `to_dataframe`. Defaults to None.
            kwargs (dict, optional): Storage options with respect to output storage type.
                    for example:
                    key, secret, endpoint for oss,
                    key, secret, client_kwargs for s3,
                    host, port for hdfs,
                    None for local.

        Returns:
            :class:`graphscope.framework.context.ResultDAGNode`, evaluated in eager mode.
        """
        protocol = fd.split("://")[0]
        # Still use the stream to write to file,
        # as the C++ adaptor in Vineyard requires arrow >= 4.0.0
        if protocol in ("file", "hdfs", "hive", "oss", "s3"):
            df = self.to_vineyard_dataframe(selector, vertex_range)
            op = dag_utils.to_data_sink(df, fd, **kwargs)
        else:
            check_argument(
                isinstance(selector, Mapping), "selector of to_dataframe must be a dict"
            )
            for _, value in selector.items():
                self._check_selector(value)
            _ensure_consistent_label(self.context_type, selector)
            selector = json.dumps(selector)
            vertex_range = utils.transform_vertex_range(vertex_range)
            op = dag_utils.output(self, fd, selector, vertex_range, **kwargs)
        return ResultDAGNode(self, op)

    def __del__(self):
        try: