Return a new DStream by applying groupByKey on each RDD.
484515
def groupByKey(self, numPartitions=None):
        
        if numPartitions is None:
            numPartitions = self._sc.defaultParallelism
        return self.transform(lambda rdd: rdd.groupByKey(numPartitions))
182546
def _createFromRDD(self, rdd, schema, samplingRatio):
        
        if schema is None or isinstance(schema, (list, tuple)):
            struct = self._inferSchema(rdd, samplingRatio, names=schema)
            converter = _create_converter(struct)
            rdd = rdd.map(converter)
            if isinstance(schema, (list, tuple)):
                for i, name in enumerate(schema):
                    struct.fields[i].name = name
                    struct.names[i] = name
            schema = struct

        elif not isinstance(schema, StructType):
            raise TypeError("schema should be StructType or list or None, but got: %s" % schema)

        # convert python objects to sql data
        rdd = rdd.map(schema.toInternal)
        return rdd, schema
1107476
def create(cls, train_ds, valid_ds, test_ds=None, path:PathOrStr='.', no_check:bool=False, bs=64, val_bs:int=None,
               num_workers:int=0, device:torch.device=None, collate_fn:Callable=data_collate,
               dl_tfms:Optional[Collection[Callable]]=None, bptt:int=70, backwards:bool=False, **dl_kwargs) -> DataBunch:
        ""
        datasets = cls._init_ds(train_ds, valid_ds, test_ds)
        val_bs = ifnone(val_bs, bs)
        datasets = [LanguageModelPreLoader(ds, shuffle=(i==0), bs=(bs if i==0 else val_bs), bptt=bptt, backwards=backwards)
                    for i,ds in enumerate(datasets)]
        val_bs = bs
        dls = [DataLoader(d, b, shuffle=False, **dl_kwargs) for d,b in zip(datasets, (bs,val_bs,val_bs,val_bs)) if d is not None]
        return cls(*dls, path=path, device=device, dl_tfms=dl_tfms, collate_fn=collate_fn, no_check=no_check)
485488
public CassandraJavaRDD<R> toEmptyCassandraRDD() {
        CassandraRDD<R> newRDD = rdd().toEmptyCassandraRDD();
        return wrap(newRDD);
    }
1359776
def drop_streams(streams, kdims, keys):
    
    stream_params = stream_parameters(streams)
    inds, dims = zip(*[(ind, kdim) for ind, kdim in enumerate(kdims)
                       if kdim not in stream_params])
    get = operator.itemgetter(*inds) # itemgetter used for performance
    keys = (get(k) for k in keys)
    return dims, ([wrap_tuple(k) for k in keys] if len(inds) == 1 else list(keys))
1157734
def dump_stream(self, iterator, stream):
        
        batches = (self._create_batch(series) for series in iterator)
        super(ArrowStreamPandasSerializer, self).dump_stream(batches, stream)
301848
def to_simple_rdd(sc, features, labels):
    
    pairs = [(x, y) for x, y in zip(features, labels)]
    return sc.parallelize(pairs)
287593
def _use_aggregation(self, agg, columns=None):
        """"""
        if not columns:
            columns = self._columns
        from pyspark.sql import functions as F
        aggs = map(lambda column: agg(column).alias(column), self._columns)
        aggRdd = self._grouped_spark_sql.agg(*aggs)
        df = DataFrame.from_schema_rdd(aggRdd, self._by)
        return df
467694
def from_rdd_of_dataframes(self, rdd, column_idxs=None):
        """"""
        def frame_to_spark_sql(frame):
            """Convert a Panda's DataFrame into Spark SQL Rows"""
            return [r.tolist() for r in frame.to_records()]

        def frame_to_schema_and_idx_names(frames):
            """Returns the schema and index names of the frames. Useful
            if the frame is large and we wish to avoid transfering
            the entire frame. Only bothers to apply once per partiton"""
            try:
                frame = frames.next()
                return [(list(frame.columns), list(frame.index.names))]
            except StopIteration:
                return []

        # Store if the RDD was persisted so we don't uncache an
        # explicitly cached input.
        was_persisted = rdd.is_cached
        # If we haven't been supplied with the schema info cache the RDD
        # since we are going to eveluate the first partition and then eveluate
        # the entire RDD as part of creating a Spark DataFrame.
        (schema, index_names) = ([], [])
        if not column_idxs:
            rdd.cache()
            (schema, index_names) = rdd.mapPartitions(
                frame_to_schema_and_idx_names).first()
        else:
            (schema, index_names) = column_idxs
        # Add the index_names to the schema.
        index_names = _normalize_index_names(index_names)
        schema = index_names + schema
        ddf = DataFrame.from_schema_rdd(
            self.sql_ctx.createDataFrame(rdd.flatMap(frame_to_spark_sql),
                                         schema=schema))
        ddf._index_names = index_names
        if not was_persisted:
            rdd.unpersist()
        return ddf
1038247
def groupby(self, by=None, axis=0, level=None, as_index=True, sort=True,
                group_keys=True, squeeze=False):
        """"""
        from sparklingpandas.groupby import GroupBy
        return GroupBy(self, by=by, axis=axis, level=level, as_index=as_index,
                       sort=sort, group_keys=group_keys, squeeze=squeeze)
657569
public static <T> Stream<T> buildReversedStream(Collection<T> collection) {
        return JMCollections.getReversed(collection).stream();
    }
745645
def trainOn(self, dstream):
        """"""
        self._validate(dstream)

        def update(rdd):
            self._model.update(rdd, self._decayFactor, self._timeUnit)

        dstream.foreachRDD(update)
661761
def time_series_rdd_from_pandas_series_rdd(series_rdd):
    
    first = series_rdd.first()
    dt_index = irregular(first[1].index, series_rdd.ctx)
    return TimeSeriesRDD(dt_index, series_rdd.mapValues(lambda x: x.values))
985426
def coalesce(num_partitions)
      if self.is_a?(PipelinedRDD)
        deser = @command.serializer
      else
        deser = @command.deserializer
      end

      new_jrdd = jrdd.coalesce(num_partitions)
      RDD.new(new_jrdd, context, @command.serializer, deser)
    end
229619
@SuppressWarnings("unchecked")
    public CassandraJavaPairRDD<K, V> withAscOrder() {
        CassandraRDD<Tuple2<K, V>> newRDD = rdd().withAscOrder();
        return wrap(newRDD);
    }
770164
def add_from_datastream(self, source_ds):
        ''''''
        if source_ds.name != self.name:
            raise DataStreamError("Data from a different stream 'f{source_ds.name}' cannot be add to this stream")
        if self._data_runs is None:
            raise DataStreamError("Cannot add data to a resident datastream.")

        if self.cluster_count < source_ds.cluster_count:
            self.cluster_count = source_ds.cluster_count
        if self.size == 0 and source_ds.size:
            self.size = source_ds.size
            self.alloc_size = source_ds.alloc_size
        if source_ds._data_runs:
            self._data_runs += source_ds._data_runs
            self._data_runs_sorted = False
1268013
def rdd(self):
        
        if self._lazy_rdd is None:
            jrdd = self._jdf.javaToPython()
            self._lazy_rdd = RDD(jrdd, self.sql_ctx._sc, BatchedSerializer(PickleSerializer()))
        return self._lazy_rdd