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)
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.