blob: 5de7057a8872b3c92b1abf69dfe22c7456909c53 [file] [log] [blame]
#!/usr/bin/env python3
# -*- encoding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
""" module for defining Heron topologies in Python
import os
import uuid
import heronpy.api.api_constants as api_constants
import six
from heronpy.api.component.component_spec import HeronComponentSpec
from heronpy.api.serializer import default_serializer
from heronpy.proto import topology_pb2
class TopologyType(type):
"""Metaclass to define a Heron topology in Python"""
api_constants.TOPOLOGY_DEBUG: "false",
api_constants.TOPOLOGY_STMGRS: "1",
api_constants.TOPOLOGY_MAX_SPOUT_PENDING: "100",
api_constants.TOPOLOGY_RELIABILITY_MODE: api_constants.TopologyReliabilityMode.ATMOST_ONCE,
def __new__(mcs, classname, bases, class_dict):
bolt_specs = {}
spout_specs = {}
# Copy HeronComponentSpec items out of class_dict
specs = TopologyType.class_dict_to_specs(class_dict)
for spec in iter(list(specs.values())):
if spec.is_spout:
TopologyType.add_spout_specs(spec, spout_specs)
TopologyType.add_bolt_specs(spec, bolt_specs)
if classname != 'Topology' and not spout_specs:
raise ValueError("A Topology requires at least one Spout")
topology_config = TopologyType.class_dict_to_topo_config(class_dict)
if classname != 'Topology':
class_dict['_topo_config'] = topology_config
class_dict['_protobuf_bolts'] = bolt_specs
class_dict['_protobuf_spouts'] = spout_specs
class_dict['_heron_specs'] = list(specs.values())
# create topology protobuf here
TopologyType.init_topology(classname, class_dict)
return type.__new__(mcs, classname, bases, class_dict)
def class_dict_to_specs(mcs, class_dict):
"""Takes a class `__dict__` and returns `HeronComponentSpec` entries"""
specs = {}
for name, spec in list(class_dict.items()):
if isinstance(spec, HeronComponentSpec):
# Use the variable name as the specification name.
if is None: = name
if in specs:
raise ValueError("Duplicate component name: %s" %
specs[] = spec
return specs
def class_dict_to_topo_config(mcs, class_dict):
Takes a class `__dict__` and returns a map containing topology-wide
The returned dictionary is a sanitized `dict` of type `<str ->
This classmethod firsts insert default topology configuration and then
overrides it with a given topology-wide configuration. Note that this
configuration will be overriden by a component-specific configuration at
topo_config = {}
# add defaults
for name, custom_config in list(class_dict.items()):
if name == 'config' and isinstance(custom_config, dict):
sanitized_dict = mcs._sanitize_config(custom_config)
return topo_config
def add_spout_specs(mcs, spec, spout_specs):
if not spec.outputs:
raise ValueError(
"%s: %s requires at least one output, because it is a spout" %
spout_specs[] = spec.get_protobuf()
def add_bolt_specs(mcs, spec, bolt_specs):
if not spec.inputs:
raise ValueError(
"%s: %s requires at least one input, because it is a bolt" %
bolt_specs[] = spec.get_protobuf()
def get_topology_config_protobuf(mcs, class_dict):
config = topology_pb2.Config()
conf_dict = class_dict['_topo_config']
for key, value in list(conf_dict.items()):
if isinstance(value, str):
kvs = config.kvs.add()
kvs.key = key
kvs.value = value
kvs.type = topology_pb2.ConfigValueType.Value("STRING_VALUE")
# need to serialize
kvs = config.kvs.add()
kvs.key = key
kvs.serialized_value = default_serializer.serialize(value)
kvs.type = topology_pb2.ConfigValueType.Value("PYTHON_SERIALIZED_VALUE")
return config
def init_topology(mcs, classname, class_dict):
"""Initializes a topology protobuf"""
if classname == 'Topology':
# Base class can't initialize protobuf
heron_options = TopologyType.get_heron_options_from_env()
initial_state = heron_options.get("cmdline.topology.initial.state", "RUNNING")
tmp_directory = heron_options.get("cmdline.topologydefn.tmpdirectory")
if tmp_directory is None:
raise RuntimeError("Topology definition temp directory not specified")
topology_name = heron_options.get("", classname)
topology_id = topology_name + str(uuid.uuid4())
# create protobuf
topology = topology_pb2.Topology() = topology_id = topology_name
topology.state = topology_pb2.TopologyState.Value(initial_state)
TopologyType.add_bolts_and_spouts(topology, class_dict)
class_dict['topology_name'] = topology_name
class_dict['topology_id'] = topology_id
class_dict['protobuf_topology'] = topology
class_dict['topologydefn_tmpdir'] = tmp_directory
class_dict['heron_runtime_options'] = heron_options
def get_heron_options_from_env():
"""Retrieves heron options from the `HERON_OPTIONS` environment variable.
Heron options have the following format:
In this case, the returned map will contain:
"cmdline.topologydefn.tmpdirectory": "/var/folders/tmpdir",
"cmdline.topology.initial.state": "PAUSED"
Currently supports the following options natively:
- `cmdline.topologydefn.tmpdirectory`: (required) the directory to which this
topology's defn file is written
- `cmdline.topology.initial.state`: (default: "RUNNING") the initial state of the topology
- ``: (default: class name) topology name on deployment
Returns: map mapping from key to value
heron_options_raw = os.environ.get("HERON_OPTIONS")
if heron_options_raw is None:
raise RuntimeError("HERON_OPTIONS environment variable not found")
options = {}
for option_line in heron_options_raw.replace("%%%%", " ").split(','):
key, sep, value = option_line.partition("=")
if sep:
options[key] = value
raise ValueError("Invalid HERON_OPTIONS part %r" % option_line)
return options
def add_bolts_and_spouts(mcs, topology, class_dict):
spouts = list(class_dict["_protobuf_spouts"].values())
bolts = list(class_dict["_protobuf_bolts"].values())
for spout in spouts:
added = topology.spouts.add()
for bolt in bolts:
added = topology.bolts.add()
def _sanitize_config(custom_config):
"""Checks whether a given custom_config and returns a sanitized dict <str -> (str|object)>
It checks if keys are all strings and sanitizes values of a given dictionary as follows:
- If string, number or boolean is given as a value, it is converted to string.
For string and number (int, float), it is converted to string by a built-in ``str()`` method.
For a boolean value, ``True`` is converted to "true" instead of "True", and ``False`` is
converted to "false" instead of "False", in order to keep the consistency with
Java configuration.
- If neither of the above is given as a value, it is inserted into the sanitized dict as it is.
These values will need to be serialized before adding to a protobuf message.
sanitized = {}
for key, value in list(custom_config.items()):
if not isinstance(key, str):
raise TypeError("Key for topology-wide configuration must be string, given: %s: %s"
% (str(type(key)), str(key)))
if isinstance(value, bool):
sanitized[key] = "true" if value else "false"
elif isinstance(value, (str, int, float)):
sanitized[key] = str(value)
sanitized[key] = value
return sanitized
class Topology(object):
"""Topology is an abstract class for defining a topology
Topology writers can define their custom topology by inheriting this class.
The usage of this class is compatible with StreamParse API.
Defining a topology is simple. Topology writers need to create a subclass, in which information
about the components in their topology and how they connect to each other are specified
by placing `HeronComponentSpec` as class instances.
For more information, refer to `spec()` method of both the `Bolt` and `Spout` classes.
In addition, you can also set a topology-wide configuration, by adding a ``config`` class
attribute to your topology class, that is dict mapping from option names to their values.
Note that topology-wide configurations are overridden by component-specific configurations
that might be specified from ``spec()`` method of ``Bolt`` or ``Spout`` class.
**Example**: A sample `WordCountTopology` can be defined as follows:
from heronpy.api.src.python import Topology
from heronpy.examples.src.python import WordSpout, CountBolt
class WordCount(Topology):
config = {"topology.wide.config": "some value"}
word_spout = WordSpout.spec(par=1)
count_bolt = CountBolt.spec(par=1,
inputs={word_spout: Grouping.fields('word')},
config={"count_bolt.specific.config": "another value"})
# pylint: disable=no-member
def write(cls):
"""Writes the Topology .defn file to ``dest``
This classmethod is meant be used by heron-cli when submitting a topology.
if cls.__name__ == 'Topology':
raise ValueError("The base Topology class cannot be writable")
filename = "%s.defn" % cls.topology_name
path = os.path.join(cls.topologydefn_tmpdir, filename)
with open(path, 'wb') as f:
class TopologyBuilder(object):
"""Builder for heronpy.api.src.python topology
This class dynamically creates a subclass of `Topology` with given spouts and
bolts and writes its definition files when `build_and_submit()` is called.
**Example**: A sample `WordCountTopology` can be defined as follows:
import sys
from heronpy.api.src.python import TopologyBuilder
from heronpy.examples.spout import WordSpout
from heronpy.examples.bolt import CountBolt
if __name__ == '__main__':
builder = TopologyBuilder(name=sys.argv[1])
word_spout = builder.add_spout("word-spout", WordSpout, 2)
builder.add_bolt("count-bolt", CountBolt, 2,
inputs={word_spout: Grouping.fields('word')},
config={"count_bolt.specific.config": "some value"})
def __init__(self, name):
"""Initialize this TopologyBuilder
:type name: str
:param name: topology name
assert name is not None and isinstance(name, str) and name != "Topology"
self.topology_name = name
self._specs = {}
self._topology_config = {}
def add_spec(self, *specs):
"""Add specs to the topology
:type specs: HeronComponentSpec
:param specs: specs to add to the topology
for spec in specs:
if not isinstance(spec, HeronComponentSpec):
raise TypeError("Argument to add_spec needs to be HeronComponentSpec, given: %s"
% str(spec))
if is None:
raise ValueError("TopologyBuilder cannot take a spec without name")
if == "config":
raise ValueError("config is a reserved name")
if in self._specs:
raise ValueError("Attempting to add duplicate spec name: %r %r" % (, spec))
self._specs[] = spec
def add_spout(self, name, spout_cls, par, config=None, optional_outputs=None):
"""Add a spout to the topology"""
spout_spec = spout_cls.spec(name=name, par=par, config=config,
return spout_spec
def add_bolt(self, name, bolt_cls, par, inputs, config=None, optional_outputs=None):
"""Add a bolt to the topology"""
bolt_spec = bolt_cls.spec(name=name, par=par, inputs=inputs, config=config,
return bolt_spec
def set_config(self, config):
"""Set topology-wide configuration to the topology
:type config: dict
:param config: topology-wide config
if not isinstance(config, dict):
raise TypeError("Argument to set_config needs to be dict, given: %s" % str(config))
self._topology_config = config
def _construct_topo_class_dict(self):
class_dict = self._specs.copy()
class_dict["config"] = self._topology_config
return class_dict
def build_and_submit(self):
"""Builds the topology and submits to the destination"""
class_dict = self._construct_topo_class_dict()
topo_cls = TopologyType(self.topology_name, (Topology,), class_dict)