Return a new DStream where each RDD is transformed by ``f``.

        :param f: transformation function
        :rtype: DStream

        The transformation function can have arguments ``(time, rdd_a, rdd_b)``
        or ``(rdd_a, rdd_b)``.
492038
def transformWith(self, func, other, keepSerializer=False):
        
        if func.__code__.co_argcount == 2:
            two_arg_func = func
            func = lambda _, rdd_a, rdd_b: two_arg_func(rdd_a, rdd_b)

        return TransformedWithDStream(self, func, other)
199557
@Override protected Frame transformImpl(Frame f) {
    final double[] fmeans = means;
    final double[] fmults = ArrayUtils.invert(sdevs);
    return new MRTask() {
      @Override public void map(Chunk[] cs, NewChunk[] ncs) {
        double[] in = new double[cs.length];
        for(int row=0; row<cs[0]._len; row++) {
          for(int col=0; col<cs.length; col++)
            in[col] = cs[col].atd(row);
          GenMunger.scaleInPlace(fmeans, fmults, in);
          for(int col=0; col<ncs.length; col++)
            ncs[col].addNum(in[col]);
        }
      }
    }.doAll(f.numCols(), Vec.T_NUM, f).outputFrame(f.names(), f.domains());
  }
1036895
def partition_t(wrap, func):
    
    return Transformation(
        'partition({0})'.format(name(func)),
        partial(partition_impl, wrap, func),
        None
    )
464097
def union(self, *dstreams):
        
        if not dstreams:
            raise ValueError("should have at least one DStream to union")
        if len(dstreams) == 1:
            return dstreams[0]
        if len(set(s._jrdd_deserializer for s in dstreams)) > 1:
            raise ValueError("All DStreams should have same serializer")
        if len(set(s._slideDuration for s in dstreams)) > 1:
            raise ValueError("All DStreams should have same slide duration")
        cls = SparkContext._jvm.org.apache.spark.streaming.api.java.JavaDStream
        jdstreams = SparkContext._gateway.new_array(cls, len(dstreams))
        for i in range(0, len(dstreams)):
            jdstreams[i] = dstreams[i]._jdstream
        return DStream(self._jssc.union(jdstreams), self, dstreams[0]._jrdd_deserializer)
882773
public static <R> Stream<R> zip(final FloatStream a, final FloatStream b, final FloatStream c, final FloatTriFunction<R> zipFunction) {
        return zip(a.iteratorEx(), b.iteratorEx(), c.iteratorEx(), zipFunction).onClose(newCloseHandler(N.asList(a, b, c)));
    }
328950
def fcat(*fs):
    
    items = list()
    for f in fs:
        if isinstance(f, boolfunc.Function):
            items.append(f)
        elif isinstance(f, farray):
            items.extend(f.flat)
        else:
            raise TypeError("expected Function or farray")
    return farray(items)
150403
def multi_ops(data_stream, *funcs):
    

    assert all(callable(func) for func in funcs), 'multi_ops can only apply functions to the first argument'
    assert len(funcs), 'multi_ops needs at least one function to apply to data_stream'

    for i in data_stream:
        if len(funcs) > 1:
            yield tuple(func(i) for func in funcs)
        elif len(funcs) == 1:
            yield funcs[0](i)
53613
def fdfilter(data, *filt, **kwargs):
    
    # parse keyword args
    inplace = kwargs.pop('inplace', False)
    analog = kwargs.pop('analog', False)
    fs = kwargs.pop('sample_rate', None)
    if kwargs:
        raise TypeError("filter() got an unexpected keyword argument '%s'"
                        % list(kwargs.keys())[0])

    # parse filter
    if fs is None:
        fs = 2 * (data.shape[-1] * data.df).to('Hz').value
    form, filt = parse_filter(filt, analog=analog, sample_rate=fs)
    lti = signal.lti(*filt)

    # generate frequency response
    freqs = data.frequencies.value.copy()
    fresp = numpy.nan_to_num(abs(lti.freqresp(w=freqs)[1]))

    # apply to array
    if inplace:
        data *= fresp
        return data
    new = data * fresp
    return new
735480
public <T2, R> ValueTransformer<W,R> zip(Iterable<? extends T2> iterable,
            BiFunction<? super T, ? super T2, ? extends R> fn) {
        return this.unitAnyM(this.transformerStream().map(v->v.zip(iterable,fn)));
    }
752244
def partitionBy(self, numPartitions, partitionFunc=portable_hash):
        
        return self.transform(lambda rdd: rdd.partitionBy(numPartitions, partitionFunc))
1166854
def _pack_fn(self):
    
    if not self.packed_length:
      return None
    def my_fn(records):
      """Function from list of TFRecords to list of TFRecords."""
      examples = []
      for record in records:
        x = tf.train.Example()
        x.ParseFromString(record)
        example_dict = {}
        if self.has_inputs:
          example_dict["inputs"] = [
              int(i) for i in x.features.feature["inputs"].int64_list.value]
        example_dict["targets"] = [
            int(i) for i in x.features.feature["targets"].int64_list.value]
        examples.append(example_dict)
      examples = list(self._maybe_pack_examples(examples))
      return [
          generator_utils.to_example(x).SerializeToString() for x in examples]
    return my_fn
1196183
public <V> Resources<V> transform(Function<T, V> fn) {
        return new Resources<>(
                resources.stream().map(fn).collect(Collectors.toList()),
                get_self(),
                pagination,
                Object.class
        ).withLinks(getLinks());
    }
669703
function (f, d) {

        f = makeCompareFunction(f, d);
        f.thenBy = tb;
        return f;
    }
854964
def mfds2multimfd(mfds):
    
    _, kind = mfds[0].tag.split('}')
    node = Node('multiMFD', dict(kind=kind, size=len(mfds)))
    lengths = None
    for field in mfd.multi_mfd.ASSOC[kind][1:]:
        alias = mfd.multi_mfd.ALIAS.get(field, field)
        if field in ('magnitudes', 'occurRates'):
            data = [~getattr(m, field) for m in mfds]
            lengths = [len(d) for d in data]
            data = sum(data, [])  # list of lists
        else:
            try:
                data = [m[alias] for m in mfds]
            except KeyError:
                if alias == 'binWidth':
                    # missing bindWidth in GR MDFs is ok
                    continue
                else:
                    raise
        node.append(Node(field, text=collapse(data)))
        if lengths:  # this is the last field if present
            node.append(Node('lengths', text=collapse(lengths)))
    return node
373568
def SpiceUDREFN(f):
    
    @functools.wraps(f)
    def wrapping_udrefn(t1, t2, s1, s2, t):
        result = f(t1, t2, s1, s2)
        t[0] = c_double(result)

    return UDREFN(wrapping_udrefn)
618646
def registerFunction(self, name, f, returnType=None):
        
        warnings.warn(
            "Deprecated in 2.3.0. Use spark.udf.register instead.",
            DeprecationWarning)
        return self.sparkSession.udf.register(name, f, returnType)
846414
def map(self, f):
        
        return self._replace(recver=self.recver.map(f))