这样做可以使所得数组的形状稍微小于所需的最大值,或者使得它们恰好具有所需的最大值,除了末尾的一些剩余部分。
基本逻辑是计算分割数组的参数,然后使用array_split https://docs.scipy.org/doc/numpy-1.14.0/reference/generated/numpy.array_split.html沿数组的每个轴(或维度)分割数组。
我们需要numpy
and math
模块和示例数组:
import math
import numpy
a = numpy.random.choice([1,2,3,4], (1009,1009))
略小于最大值
逻辑
首先将最终块大小的形状沿着您想要将其分割成元组的每个维度存储:
chunk_shape = (50, 50)
array_split
一次仅沿一个轴(或维度)或一个数组进行分割。因此,让我们从第一个轴开始。
-
计算我们需要将数组分割成的部分的数量:
num_sections = math.ceil(a.shape[0] / chunk_shape[0])
在我们的示例中,这是 21 (1009 / 50 = 20.18
).
-
现在将其分开:
first_split = numpy.array_split(a, num_sections, axis=0)
这为我们提供了 21 个(请求的部分的数量)numpy 数组的列表,这些数组被分割,因此它们在第一维中不大于 50:
print(len(first_split))
# 21
print({i.shape for i in first_split})
# {(48, 1009), (49, 1009)}
# These are the distinct shapes, so we don't see all 21 separately
在本例中,它们沿着该轴分别为 48 和 49。
-
我们可以对第二个维度的每个新数组执行相同的操作:
num_sections = math.ceil(a.shape[1] / chunk_shape[1])
second_split = [numpy.array_split(a2, num_sections, axis=1) for a2 in first_split]
这给了我们一个列表的列表。每个子列表包含我们想要的大小的 numpy 数组:
print(len(second_split))
# 21
print({len(i) for i in second_split})
# {21}
# All sublists are 21 long
print({i2.shape for i in second_split for i2 in i})
# {(48, 49), (49, 48), (48, 48), (49, 49)}
# Distinct shapes
功能齐全
我们可以使用递归函数实现任意维度:
def split_to_approx_shape(a, chunk_shape, start_axis=0):
if len(chunk_shape) != len(a.shape):
raise ValueError('chunk length does not match array number of axes')
if start_axis == len(a.shape):
return a
num_sections = math.ceil(a.shape[start_axis] / chunk_shape[start_axis])
split = numpy.array_split(a, num_sections, axis=start_axis)
return [split_to_approx_shape(split_a, chunk_shape, start_axis + 1) for split_a in split]
我们这样称呼它:
full_split = split_to_approx_shape(a, (50,50))
print({i2.shape for i in full_split for i2 in i})
# {(48, 49), (49, 48), (48, 48), (49, 49)}
# Distinct shapes
精确形状加上余数
逻辑
如果我们想要更花哨一点并且让所有新数组都是exactly除了尾部剩余数组之外的指定大小,我们可以通过传递要分割的索引列表来做到这一点array_split
.
-
首先建立索引数组:
axis = 0
split_indices = [chunk_shape[axis]*(i+1) for i in range(math.floor(a.shape[axis] / chunk_shape[axis]))]
这给出了一个索引列表,每个索引从最后一个开始:
print(split_indices)
# [50, 100, 150, 200, 250, 300, 350, 400, 450, 500, 550, 600, 650, 700, 750, 800, 850, 900, 950, 1000]
-
然后拆分:
first_split = numpy.array_split(a, split_indices, axis=0)
print(len(first_split))
# 21
print({i.shape for i in first_split})
# {(9, 1009), (50, 1009)}
# Distinct shapes, so we don't see all 21 separately
print((first_split[0].shape, first_split[1].shape, '...', first_split[-2].shape, first_split[-1].shape))
# ((50, 1009), (50, 1009), '...', (50, 1009), (9, 1009))
-
然后再次对于第二个轴:
axis = 1
split_indices = [chunk_shape[axis]*(i+1) for i in range(math.floor(a.shape[axis] / chunk_shape[axis]))]
second_split = [numpy.array_split(a2, split_indices, axis=1) for a2 in first_split]
print({i2.shape for i in second_split for i2 in i})
# {(9, 50), (9, 9), (50, 9), (50, 50)}
功能齐全
调整递归函数:
def split_to_shape(a, chunk_shape, start_axis=0):
if len(chunk_shape) != len(a.shape):
raise ValueError('chunk length does not match array number of axes')
if start_axis == len(a.shape):
return a
split_indices = [
chunk_shape[start_axis]*(i+1)
for i in range(math.floor(a.shape[start_axis] / chunk_shape[start_axis]))
]
split = numpy.array_split(a, split_indices, axis=start_axis)
return [split_to_shape(split_a, chunk_shape, start_axis + 1) for split_a in split]
我们以完全相同的方式称呼它:
full_split = split_to_shape(a, (50,50))
print({i2.shape for i in full_split for i2 in i})
# {(9, 50), (9, 9), (50, 9), (50, 50)}
# Distinct shapes
额外说明
表现
这些功能看起来相当快。我能够使用以下任一函数在 0.05 秒内将示例数组(包含超过 140 亿个元素)拆分为 1000 x 1000 个形状块(产生超过 14000 个新数组):
print('Building test array')
a = numpy.random.randint(4, size=(55000, 250000), dtype='uint8')
chunks = (1000, 1000)
numtests = 1000
print('Running {} tests'.format(numtests))
print('split_to_approx_shape: {} seconds'.format(timeit.timeit(lambda: split_to_approx_shape(a, chunks), number=numtests) / numtests))
print('split_to_shape: {} seconds'.format(timeit.timeit(lambda: split_to_shape(a, chunks), number=numtests) / numtests))
Output:
Building test array
Running 1000 tests
split_to_approx_shape: 0.035109398348040485 seconds
split_to_shape: 0.03113800323300747 seconds
我没有测试更高维数组的速度。
形状小于最大尺寸
如果任何维度的大小小于指定的最大值,这些函数都可以正常工作。这不需要特殊的逻辑。