Toolkits > com.ibm.streamsx.topology 2.1.0 > com.ibm.streamsx.topology.python > Python Topology API > Sample application
Example code that builds and then submits a simple topology.
from streamsx.topology.topology import Topology
import streamsx.topology.context
import transform_sample_functions
topo = Topology("transform_sample")
source = topo.source(transform_sample_functions.int_strings_transform)
i1 = source.map(lambda tuple_ : int(tuple_))
i2 = i1.map(transform_sample_functions.AddNum(17))
i2.print()
streamsx.topology.context.submit("STANDALONE", topo.graph)
The source function is passed a function that returns an Iterable, in this case transform_sample_functions.int_strings_transform.
def int_strings_transform():
return ["325", "457", "9325"]
The first map function is passed a function that returns an integer converted from the string object, in this case a lambda expression.
The second map function is passed an instance of a callable class that adds 17 to the integer, in this case transform_sample_functions.AddNum(17).
class AddNum:
def __init__(self, increment):
self.increment = increment
def __call__(self, tuple):
return tuple + self.increment
When building the topology the directory com.ibm.streamsx.topology/opt/python/packages must be in $PYTHONPATH.
The sample transform_sample.py can be found under samples/python/topology/simple. After updating the PYTHONPATH, the sample can be executed using python3 transform_sample.py.
342 474 9342