Function serialize_output

src/python/streaming/controller.py:318–351

@param utfEncodeStrings - Generally we want to utf encode only strings. But for Maps we utf encode everything because on the Java side we don't know the schema for maps so we wouldn't be able to tell which fields were encoded or not.

(output, utfEncodeAllFields=False)

        return serialized_item

def serialize_output(output, utfEncodeAllFields=False):
    """
    @param utfEncodeStrings - Generally we want to utf encode only strings.  But for
        Maps we utf encode everything because on the Java side we don't know the schema
        for maps so we wouldn't be able to tell which fields were encoded or not.
    """

    output_type = type(output)

    if output is None:
        return WRAPPED_NULL_BYTE
    elif output_type == tuple:
        return (WRAPPED_TUPLE_START +
                WRAPPED_FIELD_DELIMITER.join([serialize_output(o, utfEncodeAllFields) for o in output]) +
                WRAPPED_TUPLE_END)
    elif output_type == list:
        return (WRAPPED_BAG_START +
                WRAPPED_FIELD_DELIMITER.join([wrap_tuple(o, serialize_output(o, utfEncodeAllFields)) for o in output]) +
                WRAPPED_BAG_END)
    elif output_type == dict:
        return (WRAPPED_MAP_START +
                WRAPPED_FIELD_DELIMITER.join(['%s%s%s' % (k.encode('utf-8'), MAP_KEY, serialize_output(v, True)) for k, v in output.iteritems()]) +
                WRAPPED_MAP_END)
    elif output_type == bool:
        return ("true" if output else "false")
    elif output_type == bytearray:
        return str(output)
    elif output_type == datetime:
        return output.isoformat()
    elif utfEncodeAllFields or output_type == str or output_type == unicode:
        #unicode is necessary in cases where we're encoding non-strings.
        return unicode(output).encode('utf-8')
    else:
        return str(output)

if __name__ == '__main__':
    controller = PythonStreamingController()