Skip to content

Perf pandas ingest #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
j6k4m8 merged 4 commits into from
Mar 20, 2026
Merged

Perf pandas ingest #68

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
669a0ba
perf: improve degree codepath performance to use backend shortcuts
j6k4m8 Mar 19, 2026
b7fdbbb
perf: add named node management in IGraphBackend
j6k4m8 Mar 20, 2026
b980cff
perf: optimize edge ingestion in DynamoDB and NetworkX backends
j6k4m8 Mar 20, 2026
3bfa5da
add becnh for ingests
j6k4m8 Mar 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 23 additions & 50 deletions grand/backends/_dynamodb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from typing import Collection, Hashable, Optional
import time
import concurrent.futures

import pandas as pd
import boto3
Expand All @@ -10,9 +9,6 @@


_DEFAULT_DYNAMODB_URL = "http://localhost:4566"
_N_PARALLEL_REQUESTS = 16


def _dynamo_table_exists(table_name: str, client: boto3.client):
"""
Check to see if the DynamoDB table already exists.
Expand Down Expand Up @@ -468,61 +464,38 @@ def ingest_from_edgelist_dataframe(
edge_column_names = [
c for c in edgelist.columns if c not in [source_column, target_column]
]
sources = edgelist[source_column].tolist()
targets = edgelist[target_column].tolist()
edge_metadata = (
edgelist[edge_column_names].to_dict("records")
if edge_column_names
else [{} for _ in range(len(edgelist))]
)

tic = time.time()
with self._edge_table.batch_writer() as batch_writer:

with concurrent.futures.ThreadPoolExecutor(
_N_PARALLEL_REQUESTS
) as executor:
result_futures = list(
map(
lambda i: executor.submit(
lambda: batch_writer.put_item(
Item={
self._primary_key: f"__{edgelist._get_value(i, source_column)}__{edgelist._get_value(i, target_column)}",
self._edge_source_key: edgelist._get_value(
i, source_column
),
self._edge_target_key: edgelist._get_value(
i, target_column
),
**{
col: edgelist._get_value(i, col)
for col in edge_column_names
},
}
)
),
edgelist.index,
)
for source, target, metadata in zip(sources, targets, edge_metadata):
batch_writer.put_item(
Item={
self._primary_key: f"__{source}__{target}",
self._edge_source_key: source,
self._edge_target_key: target,
**metadata,
}
)
for future in concurrent.futures.as_completed(result_futures):
future.result()

# for edge in edges:
# batch_writer.put_item(Item=edge)
edge_toc = time.time() - tic

tic = time.time()
# Construct a unique set of nodes:
nodes = edgelist[source_column].append(edgelist[target_column]).unique()
nodes = pd.unique(
pd.concat(
[edgelist[source_column], edgelist[target_column]],
ignore_index=True,
)
)
with self._node_table.batch_writer() as batch_writer:
with concurrent.futures.ThreadPoolExecutor(
_N_PARALLEL_REQUESTS
) as executor:
result_futures = list(
map(
lambda x: executor.submit(
lambda: batch_writer.put_item(
Item={self._primary_key: str(x)}
)
),
nodes,
)
)
for future in concurrent.futures.as_completed(result_futures):
future.result()
for node in nodes:
batch_writer.put_item(Item={self._primary_key: str(node)})

return {
"node_count": len(nodes),
Expand Down
72 changes: 42 additions & 30 deletions grand/backends/_igraph.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from typing import Hashable, Collection

from igraph import Graph, InternalError
from igraph import Graph
import pandas as pd

from .backend import Backend
Expand Down Expand Up @@ -35,6 +35,13 @@ def __init__(self, directed: bool = False):
"""
self._directed = directed
self._ig = Graph(directed=self._directed)
self._node_ids_by_name = {}
self._edge_ids_by_key = {}

def _edge_key(self, u: Hashable, v: Hashable):
if self._directed:
return (u, v)
return frozenset((u, v))

def ingest_from_edgelist_dataframe(
self, edgelist: pd.DataFrame, source_column: str, target_column: str
Expand Down Expand Up @@ -70,12 +77,15 @@ def add_node(self, node_name: Hashable, metadata: dict):
Hashable: The ID of this node, as inserted

"""
metadata = metadata or {}

if self.has_node(node_name):
# Update metadata
m = self._ig.vs.find(name=node_name)
m.update_attributes(metadata)
self._ig.vs[self._node_ids_by_name[node_name]].update_attributes(metadata)
return node_name

self._ig.add_vertex(name=node_name, **metadata)
self._node_ids_by_name[node_name] = self._ig.vcount() - 1
return node_name

def get_node_by_id(self, node_name: Hashable):
Expand All @@ -90,7 +100,7 @@ def get_node_by_id(self, node_name: Hashable):

"""
return _remove_name_from_attributes(
self._ig.vs.find(name=node_name).attributes()
self._ig.vs[self._node_ids_by_name[node_name]].attributes()
)

def all_nodes_as_iterable(self, include_metadata: bool = False) -> Collection:
Expand Down Expand Up @@ -122,11 +132,7 @@ def has_node(self, u: Hashable) -> bool:
Returns:
bool: True if the node exists
"""
try:
self._ig.vs.find(name=u)
return True
except:
return False
return u in self._node_ids_by_name

def add_edge(self, u: Hashable, v: Hashable, metadata: dict):
"""
Expand All @@ -144,16 +150,24 @@ def add_edge(self, u: Hashable, v: Hashable, metadata: dict):
Hashable: The edge ID, as inserted.

"""
if self.has_edge(u, v):
# Update metadata
e = self._ig.get_eid(u, v)
self._ig.es[e].update_attributes(metadata)
metadata = metadata or {}
edge_key = self._edge_key(u, v)

if edge_key in self._edge_ids_by_key:
self._ig.es[self._edge_ids_by_key[edge_key]].update_attributes(metadata)
return
if not self.has_node(u):
self.add_node(u, {})
if not self.has_node(v):
self.add_node(v, {})
return self._ig.add_edge(source=u, target=v, **metadata)

new_edge = self._ig.add_edge(
source=self._node_ids_by_name[u],
target=self._node_ids_by_name[v],
**metadata,
)
self._edge_ids_by_key[edge_key] = self._ig.ecount() - 1
return new_edge

def all_edges_as_iterable(self, include_metadata: bool = False) -> Collection:
"""
Expand All @@ -174,13 +188,7 @@ def all_edges_as_iterable(self, include_metadata: bool = False) -> Collection:
yield e.source_vertex["name"], e.target_vertex["name"]

def has_edge(self, u, v):
try:
self._ig.get_eid(u, v)
return True
except (InternalError, ValueError):
# InternalError means no such edge
# ValueError means one vertex doesn't exist
return False
return self._edge_key(u, v) in self._edge_ids_by_key

def get_edge_by_id(self, u: Hashable, v: Hashable):
"""
Expand All @@ -194,10 +202,10 @@ def get_edge_by_id(self, u: Hashable, v: Hashable):
dict: Metadata associated with this edge

"""
try:
return self._ig.es[self._ig.get_eid(u, v)].attributes()
except InternalError:
edge_id = self._edge_ids_by_key.get(self._edge_key(u, v))
if edge_id is None:
raise IndexError(f"The edge ({u}, {v}) is not in the graph.")
return self._ig.es[edge_id].attributes()

def get_node_successors(
self, u: Hashable, include_metadata: bool = False
Expand All @@ -217,13 +225,15 @@ def get_node_neighbors(
Generator

"""
u_id = self._node_ids_by_name[u]

if include_metadata:
return {
self._ig.vs[s]["name"]: self.get_edge_by_id(u, s)
for s in self._ig.successors(u)
self._ig.vs[s]["name"]: self.get_edge_by_id(u, self._ig.vs[s]["name"])
for s in self._ig.successors(u_id)
}
else:
return iter([self._ig.vs[s]["name"] for s in self._ig.successors(u)])
return iter([self._ig.vs[s]["name"] for s in self._ig.successors(u_id)])

def get_node_predecessors(
self, u: Hashable, include_metadata: bool = False
Expand All @@ -238,13 +248,15 @@ def get_node_predecessors(
Generator

"""
u_id = self._node_ids_by_name[u]

if include_metadata:
return {
self._ig.vs[s]["name"]: self.get_edge_by_id(s, u)
for s in self._ig.predecessors(u)
self._ig.vs[s]["name"]: self.get_edge_by_id(self._ig.vs[s]["name"], u)
for s in self._ig.predecessors(u_id)
}
else:
return iter([self._ig.vs[s]["name"] for s in self._ig.predecessors(u)])
return iter([self._ig.vs[s]["name"] for s in self._ig.predecessors(u_id)])

def get_node_count(self) -> int:
"""
Expand Down
33 changes: 18 additions & 15 deletions grand/backends/_networkx.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,23 +190,26 @@ def ingest_from_edgelist_dataframe(
"""

tic = time.time()
self._nx_graph.add_edges_from(
[
(
e[source_column],
e[target_column],
{
k: v
for k, v in dict(e).items()
if k not in [source_column, target_column]
},
)
for _, e in edgelist.iterrows()
]
edge_columns = [
c for c in edgelist.columns if c not in [source_column, target_column]
]
sources = edgelist[source_column].tolist()
targets = edgelist[target_column].tolist()

if edge_columns:
self._nx_graph.add_edges_from(
zip(sources, targets, edgelist[edge_columns].to_dict("records"))
)
else:
self._nx_graph.add_edges_from(zip(sources, targets))

nodes = pd.unique(
pd.concat(
[edgelist[source_column], edgelist[target_column]],
ignore_index=True,
)
)

nodes = edgelist[source_column].append(edgelist[target_column]).unique()

return {
"node_count": len(nodes),
"node_duration": 0,
Expand Down
Loading
Loading