Convert transform functions to a Stream.
351880
function pipify(task) {

   const options = task.base ? { base: task.base } : {}

   let stream = gulp.src(task.src, options)

   // This is used to abort any further transforms in case of error.
   const state = { error: false }

   for (const transform of task.transforms)
      stream = stream.pipe(transform.pipe ? transform : pluginate(transform, state))

   if (task.rename)
      stream = stream.pipe(rename(task.rename))

   if (task.dest)
      stream = stream.pipe(gulp.dest(task.dest))

   return stream

}
1081344
function transformItem(stream, item) {
    var ret;
    if (isArray(item)) {
        if (isHashArray(item)) {
            ret = transformHashArrayData(stream, item);
        } else {
            ret = transformArrayData(stream, item);
        }
    } else {
        ret = transformHashData(stream, item);
    }
    return ret;
}
231622
function streamFactory( options ) {
		if ( !arguments.length ) {
			options = {};
		}
		options = copyOptions( options );
		/**
		* FUNCTION: createStream( value )
		*	Creates a stream.
		*
		* @param {*} value - value to be appended
		* @returns {Stream} Transform stream
		*/
		return function createStream( value ) {
			return new Stream( value, options );
		};
	}
945199
def _prepare_transforms(self):
        """"""
        for offset, value in enumerate(self._config.get(config.TRANSFORMS, [])):
            self._config[config.TRANSFORMS][offset] = self._import_class(value)
1099210
def transformWith(self, func, other, keepSerializer=False):
        
        if func.__code__.co_argcount == 2:
            oldfunc = func
            func = lambda t, a, b: oldfunc(a, b)
        assert func.__code__.co_argcount == 3, "func should take two or three arguments"
        jfunc = TransformFunction(self._sc, func, self._jrdd_deserializer, other._jrdd_deserializer)
        dstream = self._sc._jvm.PythonTransformed2DStream(self._jdstream.dstream(),
                                                          other._jdstream.dstream(), jfunc)
        jrdd_serializer = self._jrdd_deserializer if keepSerializer else self._sc.serializer
        return DStream(dstream.asJavaDStream(), self._ssc, jrdd_serializer)
1072959
private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) {
		for (StreamTransformation<?> transformation: transformations) {
			transform(transformation);
		}
		return streamGraph;
	}
224336
function(stream) {
		var exportKey = function(obj, env) {
			var keyString = keyStringGenerator(obj, env);

			if(stream.constElements) {
				if(obj.___jp_originalJsonLine)
					return keyString + separator + obj.___jp_originalJsonLine;
			} else {
				if(obj.___jp_originalJsonLine)
					delete obj.___jp_originalJsonLine;
			}

			return keyString + separator + JSON.stringify(obj);
		};

		return stream.pipe(jp.map(exportKey));
	}
1129352
def transform(func, data):
    
    out = []
    for r in data:
        out.append(func(r))
    return out
7053
function through(transform, flush, objectMode) {
  const stream = new TransformStream({objectMode})
  stream._transform = transform || pass
  if (flush) stream._flush = flush
  return stream
}
1204913
def transform(self, func):
        
        if func.__code__.co_argcount == 1:
            oldfunc = func
            func = lambda t, rdd: oldfunc(rdd)
        assert func.__code__.co_argcount == 2, "func should take one or two arguments"
        return TransformedDStream(self, func)
122047
def _transform(self, data, transform, step_size):
    '''
    
    '''
    if transform=='mean':
      total = sum( k*v for k,v in data.items() )
      count = sum( data.values() )
      data = float(total)/float(count) if count>0 else 0
    elif transform=='count':
      data = sum(data.values())
    elif transform=='min':
      data = min(data.keys() or [0])
    elif transform=='max':
      data = max(data.keys() or [0])
    elif transform=='sum':
      data = sum( k*v for k,v in data.items() )
    elif transform=='rate':
      data = { k:v/float(step_size) for k,v in data.items() }
    elif callable(transform):
      data = transform(data, step_size)
    return data
383336
function jsonTransform(options) {
  return function(key, value) {
    var ret = value;
    if (options.stripNulls && (value === null || value === undefined)) {
      // prune nulls
      ret = undefined;
    } else if (options.stripSerializerType && key === 'serializerType') {
      // prune serializerType
      ret = undefined;
    }
    return ret;
  };
}
1268003
def _transform(self, *transforms):
        
        sequence = None
        for transform in transforms:
            if sequence:
                sequence = Sequence(sequence, transform=transform)
            else:
                sequence = Sequence(self, transform=transform)
        return sequence
1026561
def transform(function):
        
        def transform_fn(_, result):
            if isinstance(result, Nothing):
                return result

            lgr.debug("Transforming %r with %r", result, function)
            try:
                return function(result)
            except:
                exctype, value, tb = sys.exc_info()
                try:
                    new_exc = StyleFunctionError(function, exctype, value)
                    # Remove the "During handling ..." since we're
                    # reraising with the traceback.
                    new_exc.__cause__ = None
                    six.reraise(StyleFunctionError, new_exc, tb)
                finally:
                    # Remove circular reference.
                    # https://docs.python.org/2/library/sys.html#sys.exc_info
                    del tb
        return transform_fn
1070282
function transform(chunk, encoding, cb) {
    // Allow objects straight through
    if (_.isPlainObject(chunk)) return cb(null, chunk);

    async.eachSeries(
      tokenize(chunk.toString('utf8')),
      (output, next) => transclude.call(this, output, next),
      err => {
        if (err) inputBuffer = '';
        cb(err);
      }
    );
  }
893173
function (transformFnc) {
            var ret = [];
            transformFnc = transformFnc || function (item, index) {
                return item;
            };


            this.each(function (item, index) {
                ret.push(transformFnc(item, index));
            });

            return ret;
        }
228781
function(stream) {
		var exportKey = function(line, env) {
			var obj = JSON.parse(' ' + line);

			var keyString = keyStringGenerator(obj, env);

			return keyString + separator + line;
		};

		return stream.pipe(jp.map(exportKey));
	}