用于自定义节点/边存储的NetworkX扩展性

Question

NetworkX是否支持对存储节点、边缘和属性的位置进行定制？例如，我想尝试两个选项：

1. 使用LevelDB /京都内阁之类的东西作为后盾。
2. 使用一些分布式数据库(Neo4j甚至HBase -我只需要节点/边缘的分布式存储)作为备份存储。

NetworkX支持这些东西的扩展性点是什么？

Answer 1

我将张贴为外部存储设置NetworkX的细微差别。Kikohs指出每个字典都有工厂，这是正确的。这些都可以被重写。

对于持久化存储，唯一真正需要特别注意的字典是节点字典。

必须特别注意类似dict的实现的行为。NetworkX类中有一些代码可以更改从内存中的字典返回的值，而不会将它们设置回。

例如：

self.succ[u][v]=datadict
self.pred[v][u]=datadict

这些值将不会持久化回存储后端。为了适应这种情况，我使用了一个内存缓存，该缓存保存内存中的对象，当这些对象被逐出时，它会将它们写入底层存储。

对于内存缓存，我使用了cachetools。如需驱逐，请参阅：缓存驱逐回调或等效

对于底层存储，我使用了plyvel (https://plyvel.readthedocs.org/en/latest/)，这是LevelDB的一个Python接口。

我也给出了字典的实现。注意，代码中仍然存在错误和错误，而且它还没有经过正确的测试，但是您得到了大致的概念。

class PlyvelBatchWrite(object):
    def __init__(self, plv_dict):
        self.__batch = plv_dict._db.write_batch()
        self.__plvd = plv_dict

    def set(self, key, val):
        self.__batch.put(self.__plvd.serializer.pack(key), self.__plvd.serializer.pack(val))

    def delete(self, key):
        self.__batch.delete(self.__plvd.serializer.pack(key))

    def clear(self):
        self.__batch.clear()

    def commit(self):
        self.__batch.write()


class PlyvelDict(MutableMapping):
    def __init__(self, directory='', db=None, serializer_factory=None, cache_factory=None, **kwargs):
        self.__directory = directory
        ensure_directory(directory)
        if isinstance(db, str) or db is None:
            if db is None:
                # generate UUID
                db = str(uuid.uuid4())
            self.__db = db
            db = plyvel.DB(self.name(), **kwargs)
        else:
            self.__db = kwargs['db']
        self._db = db
        if serializer_factory:
            self.serializer = serializer_factory()
        else:
            self.serializer = None
        if cache_factory:
            self.__cache = cache_factory(self.__cache_miss, self.__cache_evict)
        else:
            self.__cache = None

    def name(self):
        full_path = os.path.join(self.__directory, self.__db)
        return full_path

    def __cache_miss(self, key):
        b_item = self._db.get(self.serializer.pack(key))
        if b_item is not None:
            return self.serializer.unpack(b_item)
        else:
            raise KeyError(key)

    def __cache_evict(self, key, val):
        self._db.put(self.serializer.pack(key), self.serializer.pack(val))

    def __copy__(self):
        return type(self)(self.__directory, self._db, type(self.serializer), type(self.__cache), db=self.__db)

    def __getitem__(self, key):
        return self.__cache[key]

    def __setitem__(self, key, value):
        if key in self.__cache:
            self.__cache[key] = value
        self.__write_to_db(key, value)

    def __write_to_db(self, key, value):
        self._db.put(self.serializer.pack(key), self.serializer.pack(value))

    def __delitem__(self, key):
        if key in self.__cache:
            del self.__cache[key]
        self._db.delete(self.serializer.pack(key))

    def __iter__(self):
        return self.iterkeys()

    def __keytransform__(self, key):
        return key

    def __len__(self):
        return self.count()

    def __del__(self):
        self.flush()
        if not self._db.closed:
            self._db.close()

    # improved methods
    def flush(self, write_to_db=False):
        if self.__cache:
            if write_to_db:
                batch = self.set_batch()
                for key, val in self.__cache.items():
                    batch.set(key, val)
                batch.commit()
            self.__cache.clear()

    def set_batch(self):
        return PlyvelBatchWrite(self)

    def iteritems(self):
        self.flush()
        for key, value in self._db.iterator(include_key=True, include_value=True):
            yield (self.serializer.unpack(key), self.serializer.unpack(value))

    def iterkeys(self):
        self.flush()
        for key in self._db.iterator(include_key=True, include_value=False):
            yield self.serializer.unpack(key)

    def itervalues(self):
        self.flush()
        for val in self._db.iterator(include_key=False, include_value=True):
            yield self.serializer.unpack(val)

    def keys(self):
        self.flush()
        # fixes default method which calls __len__
        return list(self.iterkeys())

    def values(self):
        self.flush()
        return list(self.itervalues())

    def has_key(self, key):
        return key in self

    def clear(self):
        self.flush()
        for k in self:
            del self[k]

    def count(self):
        self.flush()
        return sum(1 for key in self)

图类：

class PersistedGraph(nx.Graph):
    def __init__(self, data=None, node_dict_factory=None, adjlist_dict_factory=None, edge_attr_dict_factory=None,
                 **attr):
        if node_dict_factory:
            self.node_dict_factory = node_dict_factory
        if adjlist_dict_factory:
            self.adjlist_dict_factory = adjlist_dict_factory
        if edge_attr_dict_factory:
            self.edge_attr_dict_factory = edge_attr_dict_factory
        nx.Graph.__init__(self, data, **attr)

Answer 2

应该可以通过子类图类和提供用户定义的工厂函数来扩展networkx。这些函数可以查询数据库并在networkx使用的字典中缓存结果。

我无法从在线文档中找到这些行，但是在代码中有：

子类(高级)：

图类使用了一种数据结构.

外部dict (node_dict)保存按节点键键的邻接列表。下一个dict (adjlist)表示邻接列表，并保存由邻居键控的边缘数据。内部dict (edge_attr)表示边缘数据，并保存由属性名称键决定的边缘属性值。

这三个dict中的每一个都可以被替换为用户定义的样条对象。一般情况下，应该维护类似块的特性，但是可以添加额外的功能。若要替换其中一个dicts，请通过更改类(！)创建一个新的图类。变量控制着工厂的那种类似于骰子的结构。变量名为node_dict_factory、adjlist_dict_factory和edge_attr_dict_factory。

    node_dict_factory : function, (default: dict)
    Factory function to be used to create the outer-most dict
    in the data structure that holds adjacency lists keyed by node.
    It should require no arguments and return a dict-like object.

    adjlist_dict_factory : function, (default: dict)
    Factory function to be used to create the adjacency list
    dict which holds edge data keyed by neighbor.
    It should require no arguments and return a dict-like object

    edge_attr_dict_factory : function, (default: dict)
    Factory function to be used to create the edge attribute
    dict which holds attrbute values keyed by attribute name.
    It should require no arguments and return a dict-like object.

我不知道对networkx有任何官方扩展。