bolt/spark/chunk.py
def _chunk(self, size="150", axis=None, padding=None):
        """
        Split values of distributed array into chunks.

        Transforms an underlying pair RDD of (key, value) into
        records of the form: (key, chunk id), (chunked value).
        Here, chunk id is a tuple identifying the chunk and
        chunked value is a subset of the data from each original value,
        that has been divided along the specified dimensions.

        Parameters
        ----------
        size : str or tuple or int
            If str, the average size (in KB) of the chunks in all value dimensions.
            If int or tuple, an explicit specification of the number chunks in
            each value dimension.

        axis : tuple, optional, default=None
            One or more axes to estimate chunks for, if provided any
            other axes will use one chunk.

        padding: tuple or int, default = None
            Number of elements per dimension that will overlap with the adjacent chunk.
            If a tuple, specifies padding along each chunked dimension; if a int, same
            padding will be applied to all chunked dimensions.
        """
        if self.split == len(self.shape) and padding is None:
            self._rdd = self._rdd.map(lambda kv: (kv[0]+(0,), array(kv[1], ndmin=1)))
            self._shape = self._shape + (1,)
            self._plan = (1,)
            self._padding = array([0])
            return self

        rdd = self._rdd
        self._plan, self._padding = self.getplan(size, axis, padding)

        if any([x + y > z for x, y, z in zip(self.plan, self.padding, self.vshape)]):
            raise ValueError("Chunk sizes %s plus padding sizes %s cannot exceed value dimensions %s along any axis"
                             % (tuple(self.plan), tuple(self.padding), tuple(self.vshape)))

        if any([x > y for x, y in zip(self.padding, self.plan)]):
            raise ValueError("Padding sizes %s cannot exceed chunk sizes %s along any axis"
                             % (tuple(self.padding), tuple(self.plan)))

        slices = self.getslices(self.plan, self.padding, self.vshape)
        labels = list(product(*[list(enumerate(s)) for s in slices]))
        scheme = [list(zip(*s)) for s in labels]

        def _chunk(record):
            k, v = record[0], record[1]
            for (chk, slc) in scheme:
                if type(k) is int:
                    k = (k,)
                yield k + chk, v[slc]

        rdd = rdd.flatMap(_chunk)
        return self._constructor(rdd, shape=self.shape, split=self.split,
                                 dtype=self.dtype, plan=self.plan, padding=self.padding, ordered=self._ordered)
def _chunk(self, size="150", axis=None, padding=None):
        if self.split == len(self.shape) and padding is None:
            self._rdd = self._rdd.map(lambda kv: (kv[0]+(0,), array(kv[1], ndmin=1)))
            self._shape = self._shape + (1,)
            self._plan = (1,)
            self._padding = array([0])
            return self

        rdd = self._rdd
        self._plan, self._padding = self.getplan(size, axis, padding)

        if any([x + y > z for x, y, z in zip(self.plan, self.padding, self.vshape)]):
            raise ValueError("Chunk sizes %s plus padding sizes %s cannot exceed value dimensions %s along any axis"
                             % (tuple(self.plan), tuple(self.padding), tuple(self.vshape)))

        if any([x > y for x, y in zip(self.padding, self.plan)]):
            raise ValueError("Padding sizes %s cannot exceed chunk sizes %s along any axis"
                             % (tuple(self.padding), tuple(self.plan)))

        slices = self.getslices(self.plan, self.padding, self.vshape)
        labels = list(product(*[list(enumerate(s)) for s in slices]))
        scheme = [list(zip(*s)) for s in labels]

        def _chunk(record):
            k, v = record[0], record[1]
            for (chk, slc) in scheme:
                if type(k) is int:
                    k = (k,)
                yield k + chk, v[slc]

        rdd = rdd.flatMap(_chunk)
        return self._constructor(rdd, shape=self.shape, split=self.split,
                                 dtype=self.dtype, plan=self.plan, padding=self.padding, ordered=self._ordered)
def _chunk ( self , size = "150" , axis = None , padding = None ) : if self . split == len ( self . shape ) and padding is None : self . _rdd = self . _rdd . map ( lambda kv : ( kv [ 0 ] + ( 0 , ) , array ( kv [ 1 ] , ndmin = 1 ) ) ) self . _shape = self . _shape + ( 1 , ) self . _plan = ( 1 , ) self . _padding = array ( [ 0 ] ) return self rdd = self . _rdd self . _plan , self . _padding = self . getplan ( size , axis , padding ) if any ( [ x + y > z for x , y , z in zip ( self . plan , self . padding , self . vshape ) ] ) : raise ValueError ( "Chunk sizes %s plus padding sizes %s cannot exceed value dimensions %s along any axis" % ( tuple ( self . plan ) , tuple ( self . padding ) , tuple ( self . vshape ) ) ) if any ( [ x > y for x , y in zip ( self . padding , self . plan ) ] ) : raise ValueError ( "Padding sizes %s cannot exceed chunk sizes %s along any axis" % ( tuple ( self . padding ) , tuple ( self . plan ) ) ) slices = self . getslices ( self . plan , self . padding , self . vshape ) labels = list ( product ( * [ list ( enumerate ( s ) ) for s in slices ] ) ) scheme = [ list ( zip ( * s ) ) for s in labels ] def _chunk ( record ) : k , v = record [ 0 ] , record [ 1 ] for ( chk , slc ) in scheme : if type ( k ) is int : k = ( k , ) yield k + chk , v [ slc ] rdd = rdd . flatMap ( _chunk ) return self . _constructor ( rdd , shape = self . shape , split = self . split , dtype = self . dtype , plan = self . plan , padding = self . padding , ordered = self . _ordered )
Split values of distributed array into chunks.

        Transforms an underlying pair RDD of (key, value) into
        records of the form: (key, chunk id), (chunked value).
        Here, chunk id is a tuple identifying the chunk and
        chunked value is a subset of the data from each original value,
        that has been divided along the specified dimensions.

        Parameters
        ----------
        size : str or tuple or int
            If str, the average size (in KB) of the chunks in all value dimensions.
            If int or tuple, an explicit specification of the number chunks in
            each value dimension.

        axis : tuple, optional, default=None
            One or more axes to estimate chunks for, if provided any
            other axes will use one chunk.

        padding: tuple or int, default = None
            Number of elements per dimension that will overlap with the adjacent chunk.
            If a tuple, specifies padding along each chunked dimension; if a int, same
            padding will be applied to all chunked dimensions.
Split values of distributed array into chunks .