<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def __generate_actor(self, instance_id, operator, input, output):
 """Generates an actor that will execute a particular instance of the logical operator Attributes: instance_id (UUID):
 The id of the instance the actor will execute. operator (Operator):
 The metadata of the logical operator. input (DataInput):
 The input gate that manages input channels of the instance (see: DataInput in communication.py). input (DataOutput):
 The output gate that manages output channels of the instance (see: DataOutput in communication.py). """

        actor_id = (operator.id, instance_id)
        # Record the physical dataflow graph (for debugging purposes)
        self.__add_channel(actor_id, input, output)
        # Select actor to construct
        if operator.type == OpType.Source:
            source = operator_instance.Source.remote(actor_id, operator, input,
                                                     output)
            source.register_handle.remote(source)
            return source.start.remote()
        elif operator.type == OpType.Map:
            map = operator_instance.Map.remote(actor_id, operator, input,
                                               output)
            map.register_handle.remote(map)
            return map.start.remote()
        elif operator.type == OpType.FlatMap:
            flatmap = operator_instance.FlatMap.remote(actor_id, operator,
                                                       input, output)
            flatmap.register_handle.remote(flatmap)
            return flatmap.start.remote()
        elif operator.type == OpType.Filter:
            filter = operator_instance.Filter.remote(actor_id, operator, input,
                                                     output)
            filter.register_handle.remote(filter)
            return filter.start.remote()
        elif operator.type == OpType.Reduce:
            reduce = operator_instance.Reduce.remote(actor_id, operator, input,
                                                     output)
            reduce.register_handle.remote(reduce)
            return reduce.start.remote()
        elif operator.type == OpType.TimeWindow:
            pass
        elif operator.type == OpType.KeyBy:
            keyby = operator_instance.KeyBy.remote(actor_id, operator, input,
                                                   output)
            keyby.register_handle.remote(keyby)
            return keyby.start.remote()
        elif operator.type == OpType.Sum:
            sum = operator_instance.Reduce.remote(actor_id, operator, input,
                                                  output)
            # Register target handle at state actor
            state_actor = operator.state_actor
            if state_actor is not None:
                state_actor.register_target.remote(sum)
            # Register own handle
            sum.register_handle.remote(sum)
            return sum.start.remote()
        elif operator.type == OpType.Sink:
            pass
        elif operator.type == OpType.Inspect:
            inspect = operator_instance.Inspect.remote(actor_id, operator,
                                                       input, output)
            inspect.register_handle.remote(inspect)
            return inspect.start.remote()
        elif operator.type == OpType.ReadTextFile:
            # TODO (john): Colocate the source with the input file
            read = operator_instance.ReadTextFile.remote(
                actor_id, operator, input, output)
            read.register_handle.remote(read)
            return read.start.remote()
        else:  # TODO (john): Add support for other types of operators
            sys.exit("Unrecognized or unsupported {} operator type.".format(
                operator.type))