blob: dc00aadadc37837f26ca3137c049561c9551ce18 [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.
import uuid
from heronpy.api.serializer import default_serializer
from heronpy.api.api_constants import TOPOLOGY_COMPONENT_PARALLELISM
from heronpy.proto import topology_pb2
from import Stream, Grouping
# pylint: disable=too-many-instance-attributes
class HeronComponentSpec(object):
"""Class to specify the information and location of components in a topology
This class is generated by the ``spec()`` method of Spout and Bolt class and
specifies how this component is located in the topology and how it is connected to
the other components. This class also retains the Python class path of the component,
so pex_loader can load the component appropriately.
def __init__(self, name, python_class_path, is_spout, par,
inputs=None, outputs=None, config=None):
self._sanitize_args(name, python_class_path, is_spout, par) = name
self.python_class_path = python_class_path
self.is_spout = is_spout
self.parallelism = par
# inputs, outputs, config will be sanitized later
self.inputs = inputs
self.outputs = outputs
self.custom_config = config
# This is used for identification, especially when name is not specified by argument
# Note that ```` might not be available until it is set by TopologyType metaclass
# so this is necessary for identification purposes. Used mainly by GlobalStreamId.
self.uuid = str(uuid.uuid4())
def _sanitize_args(name, py_class_path, is_spout, par):
# name can be None at the time this spec is initialized
assert name is None or isinstance(name, str)
assert isinstance(py_class_path, str)
assert isinstance(is_spout, bool)
assert isinstance(par, int) and par > 0
def get_protobuf(self):
"""Returns protobuf message (Spout or Bolt) of this component"""
if self.is_spout:
return self._get_spout()
return self._get_bolt()
def _get_spout(self):
"""Returns Spout protobuf message"""
spout = topology_pb2.Spout()
# Add output streams
return spout
def _get_bolt(self):
"""Returns Bolt protobuf message"""
bolt = topology_pb2.Bolt()
# Add streams
return bolt
def _get_base_component(self):
"""Returns Component protobuf message"""
comp = topology_pb2.Component() =
comp.spec = topology_pb2.ComponentObjectSpec.Value("PYTHON_CLASS_NAME")
comp.class_name = self.python_class_path
return comp
def _get_comp_config(self):
"""Returns component-specific Config protobuf message
It first adds ``topology.component.parallelism``, and is overriden by
a user-defined component-specific configuration, specified by spec().
proto_config = topology_pb2.Config()
# first add parallelism
key = proto_config.kvs.add()
key.value = str(self.parallelism)
key.type = topology_pb2.ConfigValueType.Value("STRING_VALUE")
# iterate through self.custom_config
if self.custom_config is not None:
sanitized = self._sanitize_config(self.custom_config)
for key, value in list(sanitized.items()):
if isinstance(value, str):
kvs = proto_config.kvs.add()
kvs.key = key
kvs.value = value
kvs.type = topology_pb2.ConfigValueType.Value("STRING_VALUE")
# need to serialize
kvs = proto_config.kvs.add()
kvs.key = key
kvs.serialized_value = default_serializer.serialize(value)
kvs.type = topology_pb2.ConfigValueType.Value("PYTHON_SERIALIZED_VALUE")
return proto_config
def _sanitize_config(custom_config):
"""Checks whether ``custom_config`` is sane 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.
if not isinstance(custom_config, dict):
raise TypeError("Component-specific configuration must be given as a dict type, given: %s"
% str(type(custom_config)))
sanitized = {}
for key, value in list(custom_config.items()):
if not isinstance(key, str):
raise TypeError("Key for component-specific 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
def _add_in_streams(self, bolt):
"""Adds inputs to a given protobuf Bolt message"""
if self.inputs is None:
# sanitize inputs and get a map <GlobalStreamId -> Grouping>
input_dict = self._sanitize_inputs()
for global_streamid, gtype in list(input_dict.items()):
in_stream = bolt.inputs.add(),
if isinstance(gtype, Grouping.FIELDS):
# it's a field grouping
in_stream.gtype = gtype.gtype
elif isinstance(gtype, Grouping.CUSTOM):
# it's a custom grouping
in_stream.gtype = gtype.gtype
in_stream.custom_grouping_object = gtype.python_serialized
in_stream.type = topology_pb2.CustomGroupingObjectType.Value("PYTHON_OBJECT")
in_stream.gtype = gtype
# pylint: disable=too-many-branches
def _sanitize_inputs(self):
"""Sanitizes input fields and returns a map <GlobalStreamId -> Grouping>"""
ret = {}
if self.inputs is None:
if isinstance(self.inputs, dict):
# inputs are dictionary, must be either <HeronComponentSpec -> Grouping> or
# <GlobalStreamId -> Grouping>
for key, grouping in list(self.inputs.items()):
if not Grouping.is_grouping_sane(grouping):
raise ValueError('A given grouping is not supported')
if isinstance(key, HeronComponentSpec):
# use default streamid
if is None:
# should not happen as TopologyType metaclass sets name attribute
# before calling this method
raise RuntimeError("In _sanitize_inputs(): HeronComponentSpec doesn't have a name")
global_streamid = GlobalStreamId(, Stream.DEFAULT_STREAM_ID)
ret[global_streamid] = grouping
elif isinstance(key, GlobalStreamId):
ret[key] = grouping
raise ValueError("%s is not supported as a key to inputs" % str(key))
elif isinstance(self.inputs, (list, tuple)):
# inputs are lists, must be either a list of HeronComponentSpec or GlobalStreamId
# will use SHUFFLE grouping
for input_obj in self.inputs:
if isinstance(input_obj, HeronComponentSpec):
if is None:
# should not happen as TopologyType metaclass sets name attribute
# before calling this method
raise RuntimeError("In _sanitize_inputs(): HeronComponentSpec doesn't have a name")
global_streamid = GlobalStreamId(, Stream.DEFAULT_STREAM_ID)
ret[global_streamid] = Grouping.SHUFFLE
elif isinstance(input_obj, GlobalStreamId):
ret[input_obj] = Grouping.SHUFFLE
raise ValueError("%s is not supported as an input" % str(input_obj))
raise TypeError("Inputs must be a list, dict, or None, given: %s" % str(self.inputs))
return ret
def _add_out_streams(self, spbl):
"""Adds outputs to a given protobuf Bolt or Spout message"""
if self.outputs is None:
# sanitize outputs and get a map <stream_id -> out fields>
output_map = self._sanitize_outputs()
for stream_id, out_fields in list(output_map.items()):
out_stream = spbl.outputs.add(), stream_id))
def _sanitize_outputs(self):
"""Sanitizes output fields and returns a map <stream_id -> list of output fields>"""
ret = {}
if self.outputs is None:
if not isinstance(self.outputs, (list, tuple)):
raise TypeError("Argument to outputs must be either list or tuple, given: %s"
% str(type(self.outputs)))
for output in self.outputs:
if not isinstance(output, (str, Stream)):
raise TypeError("Outputs must be a list of strings or Streams, given: %s" % str(output))
if isinstance(output, str):
# it's a default stream
if Stream.DEFAULT_STREAM_ID not in ret:
ret[Stream.DEFAULT_STREAM_ID] = list()
# output is a Stream object
if output.stream_id == Stream.DEFAULT_STREAM_ID and Stream.DEFAULT_STREAM_ID in ret:
# some default stream fields are already in there
ret[output.stream_id] = output.fields
return ret
def get_out_streamids(self):
"""Returns a set of output stream ids registered for this component"""
if self.outputs is None:
return set()
if not isinstance(self.outputs, (list, tuple)):
raise TypeError("Argument to outputs must be either list or tuple, given: %s"
% str(type(self.outputs)))
ret_lst = []
for output in self.outputs:
if not isinstance(output, (str, Stream)):
raise TypeError("Outputs must be a list of strings or Streams, given: %s" % str(output))
ret_lst.append(Stream.DEFAULT_STREAM_ID if isinstance(output, str) else output.stream_id)
return set(ret_lst)
def __getitem__(self, stream_id):
"""Get GlobalStreamId for a given stream_id"""
if stream_id not in self.get_out_streamids():
raise ValueError("A given stream id does not exist on this component: %s" % stream_id)
component_id = or self
return GlobalStreamId(componentId=component_id, streamId=stream_id)
def _get_stream_id(comp_name, stream_id):
"""Returns a StreamId protobuf message"""
proto_stream_id = topology_pb2.StreamId() = stream_id
proto_stream_id.component_name = comp_name
return proto_stream_id
def _get_stream_schema(fields):
"""Returns a StreamSchema protobuf message"""
stream_schema = topology_pb2.StreamSchema()
for field in fields:
key = stream_schema.keys.add()
key.key = field
key.type = topology_pb2.Type.Value("OBJECT")
return stream_schema
class GlobalStreamId(object):
"""Wrapper class to define stream_id and its component name
Constructor method is compatible with StreamParse's GlobalStreamId class, although
the object itself is completely different, as Heron does not use Thrift.
This is mainly used for declaring input fields when defining a topology, and internally
in HeronComponentSpec.
Note that topology writers never have to create an instance of this class by themselves,
as it is created automatically.
def __init__(self, componentId, streamId):
:type componentId: str or HeronComponentSpec
:param componentId: component id from which the tuple is emitted, or HeronComponentSpec object.
:type streamId: str
:param streamId: stream id through which the tuple is transmitted
if not isinstance(componentId, (str, HeronComponentSpec)):
raise TypeError('GlobalStreamId: componentId must be either string or HeronComponentSpec')
if not isinstance(streamId, str):
raise TypeError('GlobalStreamId: streamId must be string type')
self._component_id = componentId
self.stream_id = streamId
def component_id(self):
"""Returns component_id of this GlobalStreamId
Note that if HeronComponentSpec is specified as componentId and its name is not yet
available (i.e. when ``name`` argument was not given in ``spec()`` method in Bolt or Spout),
this property returns a message with uuid. However, this is provided only for safety
with __eq__(), __str__(), and __hash__() methods, and not meant to be called explicitly
before TopologyType class finally sets the name attribute of HeronComponentSpec.
if isinstance(self._component_id, HeronComponentSpec):
if is None:
# HeronComponentSpec instance's name attribute might not be available until
# TopologyType metaclass finally sets it. This statement is to support __eq__(),
# __hash__() and __str__() methods with safety, as raising Exception is not
# appropriate this case.
return "<No name available for HeronComponentSpec yet, uuid: %s>" % self._component_id.uuid
elif isinstance(self._component_id, str):
return self._component_id
raise ValueError("Component Id for this GlobalStreamId is not properly set: <%s:%s>"
% (str(type(self._component_id)), str(self._component_id)))
def __eq__(self, other):
return hasattr(other, 'component_id') and self.component_id == other.component_id \
and hasattr(other, 'stream_id') and self.stream_id == other.stream_id
def __hash__(self):
return hash(self.__str__())
def __str__(self):
return "%s:%s" % (self.component_id, self.stream_id)