splearn/rdd.py
def block(rdd, bsize=-1, dtype=None):
    """Block an RDD

    Parameters
    ----------

    rdd : RDD
        RDD of data points to block into either numpy arrays,
        scipy sparse matrices, or pandas data frames.
        Type of data point will be automatically inferred
        and blocked accordingly.

    bsize : int, optional, default None
        Size of each block (number of elements), if None all data points
        from each partition will be combined in a block.

    Returns
    -------

    rdd : ArrayRDD or TupleRDD or DictRDD
        The transformed rdd with added functionality
    """
    try:
        entry = rdd.first()
    except IndexError:
        # empty RDD: do not block
        return rdd

    # do different kinds of block depending on the type
    if isinstance(entry, dict):
        rdd = rdd.map(lambda x: list(x.values()))
        return DictRDD(rdd, list(entry.keys()), bsize, dtype)
    elif isinstance(entry, tuple):
        return DictRDD(rdd, bsize=bsize, dtype=dtype)
    elif sp.issparse(entry):
        return SparseRDD(rdd, bsize)
    elif isinstance(entry, np.ndarray):
        return ArrayRDD(rdd, bsize)
    else:
        return BlockRDD(rdd, bsize, dtype)
def block(rdd, bsize=-1, dtype=None):
    """Block an RDD

    Parameters
    ----------

    rdd : RDD
        RDD of data points to block into either numpy arrays,
        scipy sparse matrices, or pandas data frames.
        Type of data point will be automatically inferred
        and blocked accordingly.

    bsize : int, optional, default None
        Size of each block (number of elements), if None all data points
        from each partition will be combined in a block.

    Returns
    -------

    rdd : ArrayRDD or TupleRDD or DictRDD
        The transformed rdd with added functionality
    """
    try:
        entry = rdd.first()
    except IndexError:
        # empty RDD: do not block
        return rdd

    # do different kinds of block depending on the type
    if isinstance(entry, dict):
        rdd = rdd.map(lambda x: list(x.values()))
        return DictRDD(rdd, list(entry.keys()), bsize, dtype)
    elif isinstance(entry, tuple):
        return DictRDD(rdd, bsize=bsize, dtype=dtype)
    elif sp.issparse(entry):
        return SparseRDD(rdd, bsize)
    elif isinstance(entry, np.ndarray):
        return ArrayRDD(rdd, bsize)
    else:
        return BlockRDD(rdd, bsize, dtype)
Block an RDD

    Parameters
    ----------

    rdd : RDD
        RDD of data points to block into either numpy arrays,
        scipy sparse matrices, or pandas data frames.
        Type of data point will be automatically inferred
        and blocked accordingly.

    bsize : int, optional, default None
        Size of each block (number of elements), if None all data points
        from each partition will be combined in a block.

    Returns
    -------

    rdd : ArrayRDD or TupleRDD or DictRDD
        The transformed rdd with added functionality