def thunder_decorator(func):
    """ Decorator for functions so they could get as input a thunder.Images / thunder.Series object,
    while they are expecting an rdd. Also will return the data from rdd to the appropriate type
    Assumes only one input object of type Images/Series, and up to one output object of type RDD

    :param func: function to decorate
    :return: decorated function
    """
    @wraps(func)
    def dec(*args, **kwargs):
        # find Images / Series object in args
        result = None
        args = list(args)
        image_args = list(map(lambda x: isinstance(x, td.images.Images), args))
        series_args = list(map(lambda x: isinstance(x, td.series.Series), args))
        rdd_args = list(map(lambda x: isinstance(x, RDD), args))

        # find Images / Series object in kwargs
        image_kwargs = []
        series_kwargs = []
        rdd_kwargs = []
        for key, value in iteritems(kwargs):
            if isinstance(value, td.images.Images):
                image_kwargs.append(key)
            if isinstance(value, td.series.Series):
                series_kwargs.append(key)
            if isinstance(value, RDD):
                rdd_kwargs.append(key)

        # make sure there is only one
        count = sum(image_args) + sum(series_args) + sum(rdd_args) + len(image_kwargs) + len(series_kwargs) + \
            len(rdd_kwargs)
        if count == 0:
            raise ValueError('Wrong data type, expected [RDD, Images, Series] got None')
        if count > 1:
            raise ValueError('Expecting on input argument of type Series / Images, got: %d' % count)

        # bypass for RDD
        if sum(rdd_args) or len(rdd_kwargs):
            return func(*args, **kwargs)
        image_flag = None
        # convert to rdd and send
        if sum(image_args) > 0:
            image_flag = True
            index = np.where(image_args)[0][0]
            args[index] = args[index].tordd()
            result = func(*args, **kwargs)
        if sum(series_args) > 0:
            image_flag = False
            index = np.where(series_args)[0][0]
            args[index] = args[index].tordd()
            result = func(*args, **kwargs)

        if len(image_kwargs) > 0:
            image_flag = True
            kwargs[image_kwargs[0]] = kwargs[image_kwargs[0]].tordd()
            result = func(*args, **kwargs)

        if len(series_kwargs) > 0:
            image_flag = False
            kwargs[series_kwargs[0]] = kwargs[series_kwargs[0]].tordd()
            result = func(*args, **kwargs)

        if image_flag is None:
            raise RuntimeError('Target function did not run')

        # handle output
        if not isinstance(result, tuple):
            result = (result,)
        result_len = len(result)
        rdd_index = np.where(list(map(lambda x: isinstance(x, RDD), result)))[0]

        # no RDD as output
        if len(rdd_index) == 0:
            logging.getLogger('pySparkUtils').debug('No RDDs found in output')
            if result_len == 1:
                return result[0]
            else:
                return result

        if len(rdd_index) > 1:
            raise ValueError('Expecting one RDD as output got: %d' % len(rdd_index))
        result = list(result)
        rdd_index = rdd_index[0]

        # handle type of output
        if image_flag:
            result[rdd_index] = td.images.fromrdd(result[rdd_index])
        else:
            result[rdd_index] = td.series.fromrdd(result[rdd_index])
        if result_len == 1:
            return result[0]
        else:
            return result

    return dec
Decorator for functions so they could get as input a thunder.Images / thunder.Series object,
    while they are expecting an rdd. Also will return the data from rdd to the appropriate type
    Assumes only one input object of type Images/Series, and up to one output object of type RDD

    :param func: function to decorate
    :return: decorated function