bolt/spark/chunk.py
def unchunk(self):
        """
        Convert a chunked array back into a full array with (key,value) pairs
        where key is a tuple of indices, and value is an ndarray.
        """
        plan, padding, vshape, split = self.plan, self.padding, self.vshape, self.split
        nchunks = self.getnumber(plan, vshape)
        full_shape = concatenate((nchunks, plan))
        n = len(vshape)
        perm = concatenate(list(zip(range(n), range(n, 2*n))))

        if self.uniform:
            def _unchunk(it):
                ordered = sorted(it, key=lambda kv: kv[0][split:])
                keys, values = zip(*ordered)
                yield keys[0][:split], asarray(values).reshape(full_shape).transpose(perm).reshape(vshape)
        else:
            def _unchunk(it):
                ordered = sorted(it, key=lambda kv: kv[0][split:])
                keys, values = zip(*ordered)
                k_chks = [k[split:] for k in keys]
                arr = empty(nchunks, dtype='object')
                for (i, d) in zip(k_chks, values):
                    arr[i] = d
                yield keys[0][:split], allstack(arr.tolist())

        # remove padding
        if self.padded:
            removepad = self.removepad
            rdd = self._rdd.map(lambda kv: (kv[0], removepad(kv[0][split:], kv[1], nchunks, padding, axes=range(n))))
        else:
            rdd = self._rdd

        # skip partitionBy if there is not actually any chunking
        if array_equal(self.plan, self.vshape):
           rdd = rdd.map(lambda kv: (kv[0][:split], kv[1]))
           ordered = self._ordered
        else:
            ranges = self.kshape
            npartitions = int(prod(ranges))
            if len(self.kshape) == 0:
                partitioner = lambda k: 0
            else:
                partitioner = lambda k: ravel_multi_index(k[:split], ranges)
            rdd = rdd.partitionBy(numPartitions=npartitions, partitionFunc=partitioner).mapPartitions(_unchunk)
            ordered = True

        if array_equal(self.vshape, [1]):
            rdd = rdd.mapValues(lambda v: squeeze(v))
            newshape = self.shape[:-1]
        else:
            newshape = self.shape

        return BoltArraySpark(rdd, shape=newshape, split=self._split,
                              dtype=self.dtype, ordered=ordered)
def unchunk(self):
        """
        Convert a chunked array back into a full array with (key,value) pairs
        where key is a tuple of indices, and value is an ndarray.
        """
        plan, padding, vshape, split = self.plan, self.padding, self.vshape, self.split
        nchunks = self.getnumber(plan, vshape)
        full_shape = concatenate((nchunks, plan))
        n = len(vshape)
        perm = concatenate(list(zip(range(n), range(n, 2*n))))

        if self.uniform:
            def _unchunk(it):
                ordered = sorted(it, key=lambda kv: kv[0][split:])
                keys, values = zip(*ordered)
                yield keys[0][:split], asarray(values).reshape(full_shape).transpose(perm).reshape(vshape)
        else:
            def _unchunk(it):
                ordered = sorted(it, key=lambda kv: kv[0][split:])
                keys, values = zip(*ordered)
                k_chks = [k[split:] for k in keys]
                arr = empty(nchunks, dtype='object')
                for (i, d) in zip(k_chks, values):
                    arr[i] = d
                yield keys[0][:split], allstack(arr.tolist())

        # remove padding
        if self.padded:
            removepad = self.removepad
            rdd = self._rdd.map(lambda kv: (kv[0], removepad(kv[0][split:], kv[1], nchunks, padding, axes=range(n))))
        else:
            rdd = self._rdd

        # skip partitionBy if there is not actually any chunking
        if array_equal(self.plan, self.vshape):
           rdd = rdd.map(lambda kv: (kv[0][:split], kv[1]))
           ordered = self._ordered
        else:
            ranges = self.kshape
            npartitions = int(prod(ranges))
            if len(self.kshape) == 0:
                partitioner = lambda k: 0
            else:
                partitioner = lambda k: ravel_multi_index(k[:split], ranges)
            rdd = rdd.partitionBy(numPartitions=npartitions, partitionFunc=partitioner).mapPartitions(_unchunk)
            ordered = True

        if array_equal(self.vshape, [1]):
            rdd = rdd.mapValues(lambda v: squeeze(v))
            newshape = self.shape[:-1]
        else:
            newshape = self.shape

        return BoltArraySpark(rdd, shape=newshape, split=self._split,
                              dtype=self.dtype, ordered=ordered)
def unchunk ( self ) : plan , padding , vshape , split = self . plan , self . padding , self . vshape , self . split nchunks = self . getnumber ( plan , vshape ) full_shape = concatenate ( ( nchunks , plan ) ) n = len ( vshape ) perm = concatenate ( list ( zip ( range ( n ) , range ( n , 2 * n ) ) ) ) if self . uniform : def _unchunk ( it ) : ordered = sorted ( it , key = lambda kv : kv [ 0 ] [ split : ] ) keys , values = zip ( * ordered ) yield keys [ 0 ] [ : split ] , asarray ( values ) . reshape ( full_shape ) . transpose ( perm ) . reshape ( vshape ) else : def _unchunk ( it ) : ordered = sorted ( it , key = lambda kv : kv [ 0 ] [ split : ] ) keys , values = zip ( * ordered ) k_chks = [ k [ split : ] for k in keys ] arr = empty ( nchunks , dtype = 'object' ) for ( i , d ) in zip ( k_chks , values ) : arr [ i ] = d yield keys [ 0 ] [ : split ] , allstack ( arr . tolist ( ) ) # remove padding if self . padded : removepad = self . removepad rdd = self . _rdd . map ( lambda kv : ( kv [ 0 ] , removepad ( kv [ 0 ] [ split : ] , kv [ 1 ] , nchunks , padding , axes = range ( n ) ) ) ) else : rdd = self . _rdd # skip partitionBy if there is not actually any chunking if array_equal ( self . plan , self . vshape ) : rdd = rdd . map ( lambda kv : ( kv [ 0 ] [ : split ] , kv [ 1 ] ) ) ordered = self . _ordered else : ranges = self . kshape npartitions = int ( prod ( ranges ) ) if len ( self . kshape ) == 0 : partitioner = lambda k : 0 else : partitioner = lambda k : ravel_multi_index ( k [ : split ] , ranges ) rdd = rdd . partitionBy ( numPartitions = npartitions , partitionFunc = partitioner ) . mapPartitions ( _unchunk ) ordered = True if array_equal ( self . vshape , [ 1 ] ) : rdd = rdd . mapValues ( lambda v : squeeze ( v ) ) newshape = self . shape [ : - 1 ] else : newshape = self . shape return BoltArraySpark ( rdd , shape = newshape , split = self . _split , dtype = self . dtype , ordered = ordered )
Convert a chunked array back into a full array with (key,value) pairs
        where key is a tuple of indices, and value is an ndarray.
Convert a chunked array back into a full array with ( key value ) pairs where key is a tuple of indices and value is an ndarray .