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

Question

NetworkX 是否支持自定义节点、边和属性的存储位置？例如，我想尝试 2 个选项:

1. 使用 LevelDB/Kyoto Cabinet 之类的东西作为后备存储。

2. 使用一些分布式数据库(Neo4j 甚至 HBase - 我只需要节点/边缘的分布式存储)作为后备存储。

NetworkX 有哪些可扩展点来支持此类功能？

Answer 1

我将发布为外部存储设置 NetworkX 的细微差别。 Kikohs 正确地指出每种词典都有工厂。这些可以被覆盖。

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

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

例如:

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

这些值不会保留回存储后端。为了适应这一点，我使用了内存缓存来保存内存中的对象，当它们被逐出时，它会将它们写入底层存储。

对于内存缓存，我使用了cachetools。对于驱逐请参阅:Python functools.lru_cache eviction callback or equivalent

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

下面我还给出了字典的实现。请注意，代码中仍然存在错误和错误，并且尚未经过正确测试，但您已经了解了总体思路。

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)