在python中迭代一个巨大列表的子列表的有效方法

Question

所以我需要找到一种有效的方法来遍历 python 中的大列表。

给定：整数数组和数字（子列表的长度）

约束：数组最多 100K 个元素，元素在 range(1,2**31)

任务：为每个子列表找出最大和最小数量之间的差异。打印出最大的不同。

Ex: [4,6,3,4,8,1,9], number = 3
As far as I understand I have to go through every sublist:

[4,6,3]  max - min = 6 - 3 = 3
[6,3,4]  3
[3,4,8]  5
[4,8,1]  7
[8,1,9]  8

final max = 8

所以我的解决方案是：

import time

def difference(arr, number): 
    maxDiff = 0
    i = 0
    while i+number != len(arr)+1:
        diff = max(arr[i:i+number]) - min(arr[i:i+number])

        if diff > maxDiff:
            maxDiff = diff

        i += 1

    print maxDiff


length = 2**31
arr = random.sample(xrange(length),100000)   #array wasn't given. My sample
t0 = time.clock()
difference(arr,3)
print 'It took :',time.clock() - t0

答案：

2147101251
It took : 5.174262

我也对 for 循环做了同样的事情，这会导致更糟糕的时间：

def difference(arr,d):
    maxDiff = 0
    if len(arr) == 0:
        maxDiff = 0
    elif len(arr) == 1:
        maxDiff = arr[0]
    else:
        i = 0
        while i + d != len(arr)+1:

            array = []
            for j in xrange(d):
                array.append(arr[i + j])

            diff = max(array) - min(array)

            if diff > maxDiff:
                maxDiff = diff

            i += 1
    print maxDiff

length = 2**31
arr = random.sample(xrange(length),100000)      #array wasn't given. My sample
t0 = time.clock()
difference(arr,1000)
print 'It took :',time.clock() - t0

答案：

2147331163
It took : 14.104639

我的挑战是将时间减少到 2 秒。

最有效的方法是什么？？？

根据@rchang 和@gknicker 的回答和评论，我得到了改进。我想知道我还有什么可以做的吗？

def difference(arr,d):
    window = arr[:d]
    arrayLength = len(arr)
    maxArrayDiff = max(arr) - min(arr)

    maxDiff = 0

    while d < arrayLength:

        localMax = max(window)
        if localMax > maxDiff:
            diff = localMax - min(window)

            if diff == maxArrayDiff:
                return diff
                break
            elif diff > maxDiff:
                maxDiff = diff

        window.pop(0)
        window.append(arr[d])

        d += 1

    return maxDiff


#arr = [3,4,6,15,7,2,14,8,1,6,1,2,3,10,1]
length = 2**31
arr = random.sample(xrange(length),100000)
t0 = time.clock()
print difference(arr,1000)
print 'It took :',time.clock() - t0

答案：

2147274599
It took : 2.54171

还不错。还有其他建议吗？

Answer 1

这是我解决这个问题的尝试。

我做了很多实验和测量，得出以下结论：

1. subset_length 对性能有显着影响。
2. numpy min/max 比内置函数快得多，但仅适用于大型数组，例如 50 以下，内置函数更快。
3. 这对于subset_length 的影响
   • 低于 10 您的最新版本是最快的
   • 在 10 到 50 之间，没有 numpy（尚未发布）的我的算法版本最快
   • 50 以上我的算法是最快的
   • 在 1000 时，此算法的性能比您的算法高 100 倍

---

注意array 必须是numpy.array() 并且subset_length 必须是3 或更多。

def difference_np(array, subset_length):
    assert subset_length > 2, "subset_length must be larger than 2"
    length = array.size
    total_diff = array.max()-array.min()

    current_min = array[:subset_length].min()
    current_max = array[:subset_length].max()
    max_diff = current_max - current_min
    max_diff_index = 0
    index = subset_length
    while index < length:
        i_new = index
        i_old = index-number
        index += 1     
        new = array[i_new]            
        old = array[i_old]

        # the idea here is to avoid calculating the
        #   min/max over the entire subset as much as possible,
        #   so we treat every edge case separately.
        if new < current_min:
            current_min = new
            if old == current_max:
                current_max = array[i_old+1:i_new-1].max()
        elif new > current_max:
            current_max = new
            if old == current_min:
                current_min = array[i_old+1:i_new-1].min()
        elif old == current_min:
            current_min = array[i_old+1:i_new].min()
        elif old == current_max:
            current_max = array[i_old+1:i_new].max()
        else:
            continue

        current_diff = current_max-current_min
        if current_diff > max_diff:
            max_diff = current_diff
            max_diff_index = i_old

        # shortcut-condition
        if max_diff == total_diff:
            print('shortcut at', (index-1)/(length-subset_length), '%' )
            break

    return max_diff, max_diff_index

---

我不确定快捷条件是否真的那么有效，因为它很少应用并且需要输入数组的两次完整迭代。

---

编辑

如果算法使用list.pop(0)，则存在其他改进余地。由于list 针对右手操作进行了优化，list.pop(0) 相对昂贵。对于collections.deque，存在提供快速左侧弹出的替代方案：deque.popleft()。是给整体速度带来了不小的提升。

---

这里是我的算法的非numpy collections.deque 版本：

def difference_deque(array, subset_length):
    assert subset_length > 1, "subset_length must be larger than 1"
    length = len(array)
    total_diff = max(array)-min(array)

    current_slice = collections.deque(array[:subset_length])
    current_min = min(current_slice)
    current_max = max(current_slice)
    max_diff = current_max - current_min
    max_diff_index = 0

    index = subset_length
    while index < length:
        i_new = index
        i_old = index-number
        index += 1     
        new = array[i_new]            
        old = current_slice.popleft()

        if new < current_min:
            current_min = new
            if old == current_max:
                current_max = max(current_slice)
            current_slice.append(new)
        elif new > current_max:
            current_max = new
            if old == current_min:
                current_min = min(current_slice)
            current_slice.append(new)
        elif old == current_min:
            current_slice.append(new)
            current_min = min(current_slice)
        elif old == current_max:
            current_slice.append(new)
            current_max = max(current_slice)
        else:
            current_slice.append(new)
            continue

        current_diff = current_max-current_min
        if current_diff > max_diff:
            max_diff = current_diff
            max_diff_index = i_old+1

        # shortcut-condition
        if max_diff == total_diff:
            print('shortcut at', (index-1)/(length-number), '%' )
            break

    return max_diff, max_diff_index

---

它稍微歪曲了运行时排名： - 最多 10 个您的算法（使用双端队列）是最好的 - 最多 100 我的算法（使用双端队列）是最好的 - 100 以上我的算法（使用 numpy）是最好的

Answer 2

我提出了这个优化方案，它可能会为您的首次实施节省一些时间。我没有使用切片来隔离每次迭代要考虑的数字，而是一次使用切片来初始化“窗口”。在每次迭代中，“最右边”的元素被添加到窗口中，“最左边”的元素被逐出。

import time
import random

def difference(arr, number):
  thisSlice = arr[:number-1]
  arrSize = len(arr)
  maxDiff = -1000

  while number < arrSize:

    # Put the new element onto the window's tail
    thisSlice.append(arr[number])

    thisDiff = max(thisSlice) - min(thisSlice)
    if thisDiff > maxDiff: maxDiff = thisDiff
    number += 1

    # Get rid of the "leftmost" element, we won't need it for next iteration
    thisSlice.pop(0)

  print maxDiff

if __name__ == '__main__':
    length = 2**31
    arr = random.sample(xrange(length),100000)
    t0 = time.clock()
    difference(arr, 1000)
    print 'It took :', time.clock() - t0

至少在我的笔记本电脑上，这不会低于 2 秒，但与您发布的第一个实现相比，我确实看到了一些进步。平均而言，您的第一个解决方案在我的笔记本电脑上运行时间为 4.2 到 4.3 秒。这个零碎的窗口构建版本平均运行时间在 3.5 到 3.6 秒之间。

希望对你有帮助。

Answer 3

我认为您可以使用 as_strided 魔法来使用各种 numpy 滚动窗口函数之一——比如说我刚刚从 here 偷来的那个：

def rolling_window(a, window):
    shape = a.shape[:-1] + (a.shape[-1] - window + 1, window)
    strides = a.strides + (a.strides[-1],)
    return np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)

使用您原来的difference，但使用return 而不是print，并且arr 是一个numpy 数组：

>>> w = 3
>>> %timeit old_d = difference(arr, w)
1 loops, best of 3: 718 ms per loop
>>> %timeit q = rolling_window(arr, w); ma=q.max(1);mi=q.min(1); new_d=(ma-mi).max()
100 loops, best of 3: 5.68 ms per loop

和

>>> w = 1000
>>> %timeit old_d = difference(arr, w)
1 loops, best of 3: 25.1 s per loop
>>> %timeit q = rolling_window(arr, w); ma=q.max(1);mi=q.min(1); new_d=(ma-mi).max()
1 loops, best of 3: 326 ms per loop