Topology api

Hyrrokkin Topology API

The topology API can be used to build and run topologies. To build a topology:

from tempfile import TemporaryDirectory
from hyrrokkin.api import Topology

# provide the resource path to the hyrrokkin package containing the schema.json file
textgraph_package = "hyrrokkin.example_packages.textgraph"

with TemporaryDirectory() as tmpdir:
    t = Topology(topology_folder=tmpdir, package_list=[textgraph_package])

    # create a text processing topology to analyse word frequencies with 3 nodes and 2 links

    t.add_node("n0", "textgraph:text_input_node", data={"value": b"This is some simple text"})
    t.add_node("n1", "textgraph:word_frequency_node", properties={"threshold": 1})
    t.add_node("n2", "textgraph:table_display_node")

    t.add_link("l0", "n0", "data_out", "n1", "data_in")
    t.add_link("l1", "n1", "data_out", "n2", "data_in")
    print(t)

The same topology can be expressed using a YAML file

metadata:
  name: test topology
nodes:
  n0:
    type: textgraph:text_input_node
    data:
      value: node/n0/data/value
  n1:
    type: textgraph:word_frequency_node
    properties:
      threshold: 1
  n2:
    type: textgraph:table_display_node
links:
- n0:data_out => n1:data_in
- n1:data_out => n2:data_in

...where node/n0/data/value is a file which contains b"This is some simple text"

This YAML file can then be imported using the following API calls

from hyrrokkin.api import Topology
from tempfile import TemporaryDirectory

# provide the resource path to the package containing the schema file
textgraph_package = "hyrrokkin.example_packages.textgraph"

with TemporaryDirectory() as tmpdir:
    t = Topology(topology_folder=tmpdir,package_list=[textgraph_package])
    t.import_from("topology/topology.yaml")
    print(t)

In the links section of the YAML file, where nodes have only one input or output port, the port name can be omitted:

metadata:
  name: test topology
configuration:
  ...
nodes:
  ...
links:
- n0 => n1
- n1 => n2

Saving and loading topologies to zip format

A topology including its properties and data can be saved to and loaded from a serialised zip format file, using the following API calls. Saving first:

from tempfile import TemporaryDirectory
from hyrrokkin.api.topology import Topology

# provide the resource path to the package containing the schema file
textgraph_package = "hyrrokkin.example_packages.textgraph"

with TemporaryDirectory() as tmpdir:
    t = Topology(topology_folder=tmpdir,package_list=[textgraph_package])

    t.add_node("n0", "textgraph:text_input_node", data={"value":b"this is some text"})
    t.add_node("n1", "textgraph:word_frequency_node",properties={"threshold": 1})
    t.add_node("n2", "textgraph:table_display_node")

    t.add_link("l0", "n0", "data_out", "n1", "data_in")
    t.add_link("l1", "n1", "data_out", "n2", "data_in")

    t.export_to("topology.zip")

To load from a saved topology:

from hyrrokkin.api.topology import Topology
from tempfile import TemporaryDirectory

# provide the resource path to the package containing the schema file
textgraph_package = "hyrrokkin.example_packages.textgraph"

with TemporaryDirectory() as tmpdir:
    t = Topology(topology_folder=tmpdir,package_list=[textgraph_package])
    t.import_from("topology.zip")
    print(t)

A topology can also be exported to the YAML format. Note that the exported YAML file contains node and configuration properties and contains paths to files containing any data stored by each node and configuration.

from tempfile import TemporaryDirectory

from hyrrokkin.api import Topology


textgraph_package = "hyrrokkin.example_packages.textgraph"

with TemporaryDirectory() as tmpdir:

    t = Topology(topology_folder=tmpdir,package_list=[textgraph_package])

    t.add_node("n0", "textgraph:text_input_node", data={"value": b"This is some simple text"})
    t.add_node("n1", "textgraph:word_frequency_node", properties={"threshold": 1})
    t.add_node("n2", "textgraph:table_display_node")

    t.add_link("l0", "n0", "data_out", "n1", "data_in")
    t.add_link("l1", "n1", "data_out", "n2", "data_in")
    t.export_to("test_export/topology.yaml")

For full details on the topology API, see:

• Topology API documentation