algorithm - 将对象均匀分布在多个集合上

Question

场景是有n不同大小的对象不均匀地分布在m桶上。存储桶的大小是它包含的所有对象大小的总和。现在碰巧桶的大小变化很大。

如果我想将这些对象均匀地分布在这些桶上以便每个桶的总大小大致相同，那么什么是一个好的算法？如果算法倾向于在完全均匀的传播上减少移动大小，那就太好了。

我在 Ruby 中有这个幼稚、低效和错误的解决方案。

buckets = [ [10, 4, 3, 3, 2, 1], [5, 5, 3, 2, 1], [3, 1, 1], [2] ]

avg_size = buckets.flatten.reduce(:+) / buckets.count + 1

large_buckets = buckets.take_while {|arr| arr.reduce(:+) >= avg_size}.to_a

large_buckets.each do |large|
  smallest = buckets.last

  until ((small_sum = smallest.reduce(:+)) >= avg_size)
    break if small_sum + large.last >= avg_size
    smallest << large.pop
  end

  buckets.insert(0, buckets.pop)
end

=> [[3, 1, 1, 1, 2, 3], [2, 1, 2, 3, 3], [10, 4], [5, 5]]

Answer 1

我相信这是装箱问题的变体，因此它是 NP 难的。您的答案本质上是第一次拟合递减启发式的变体，这是一个非常好的启发式。也就是说，我相信以下内容会产生更好的结果。

• 使用平衡二叉树按大小降序对每个单独的桶进行排序。
• 计算平均大小。
• 使用平衡二叉树按大小降序对大小小于平均值的桶（“太小桶”）进行排序。
• 使用平衡二叉树对大小大于平均值的存储桶（“太大的存储桶”）按最大元素的大小排序（因此具有 {9, 1} 的存储桶将排在第一位，而具有 { 8, 5} 将排在第二位）。
• Pass1：从元素最大的bucket中取出最大的元素；如果这将其大小减小到平均值以下，则替换删除的元素并从“太大桶”的平衡二叉树中删除该桶；否则将元素放在最小的桶中，并重新索引两个修改后的桶以反映新的最小桶和具有最大元素的新“太大桶”。继续迭代，直到您删除了所有“太大的存储桶”。
• Pass2：从小到大迭代“过小桶”，从最大的“过大桶”中选出最适合的元素，不使其变成“过小桶”；从最大到最小遍历剩余的“太大的桶”，从中删除最适合的元素，而不会导致它们变成“太小的桶”。对剩余的“太小桶”做同样的事情。这个变体的结果不会像更复杂的变体那样好，因为它不会将桶从“太大”转移到“太小”类别，反之亦然（因此搜索空间将更小），但这也意味着它具有更简单的停止条件（只需遍历所有“太小”

这个想法是，通过移动 Pass1 中最大的元素，您可以更轻松地更精确地匹配 Pass2 中存储桶的大小。您使用平衡二叉树，以便在删除或添加元素后快速重新索引桶或桶树，但您可以使用链表代替（平衡二叉树在最坏情况下的性能更好，但链表可能具有更好的平均情况性能）。通过在 Pass2 中执行最佳拟合而不是首次拟合，您不太可能执行无用的移动（例如，将大小为 10 的对象从比平均值大 5 的桶移动到比平均值小 5 的桶中 - 第一次拟合将盲目地执行电影，最佳拟合将查询下一个“太大的桶”以获得更大的对象，否则将删除“

Answer 2

我最终得到了这样的东西。

• 按大小降序对存储桶进行排序。
• 按大小降序对每个单独的存储桶进行排序。
• 计算平均大小。
• 迭代每个大小大于平均大小的桶。
• 将对象按大小顺序从这些桶移动到最小的桶，直到大桶小于平均大小或目标桶达到平均大小。

Ruby 代码示例

require 'pp'

def average_size(buckets)
  (buckets.flatten.reduce(:+).to_f / buckets.count + 0.5).to_i
end

def spread_evenly(buckets)
  average = average_size(buckets)
  large_buckets = buckets.take_while {|arr| arr.reduce(:+) >= average}.to_a

  large_buckets.each do |large_bucket|
    smallest_bucket = buckets.last
    smallest_size = smallest_bucket.reduce(:+)
    large_size = large_bucket.reduce(:+)

    until (smallest_size >= average)
      break if large_size <= average
      if smallest_size + large_bucket.last > average and large_size > average
        buckets.unshift buckets.pop
        smallest_bucket = buckets.last
        smallest_size = smallest_bucket.reduce(:+)
      end
      smallest_size += smallest_object = large_bucket.pop
      large_size -= smallest_object
      smallest_bucket << smallest_object
    end

    buckets.unshift buckets.pop if smallest_size >= average
  end
  buckets
end

test_buckets = [ 
  [ [10, 4, 3, 3, 2, 1], [5, 5, 3, 2, 1], [3, 1, 1], [2] ],
  [ [4, 3, 3, 2, 2, 2, 2, 1, 1], [10, 5, 3, 2, 1], [3, 3, 3], [6] ],
  [ [1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1], [1, 1] ],
  [ [10, 9, 8, 7], [6, 5, 4], [3, 2], [1] ],
]

test_buckets.each do |buckets|
  puts "Before spread with average of #{average_size(buckets)}:"
  pp buckets
  result = spread_evenly(buckets)
  puts "Result and sum of each bucket:"
  pp result
  sizes = result.map {|bucket| bucket.reduce :+}
  pp sizes
  puts
end

输出：

Before spread with average of 12:
[[10, 4, 3, 3, 2, 1], [5, 5, 3, 2, 1], [3, 1, 1], [2]]
Result and sum of each bucket:
[[3, 1, 1, 4, 1, 2], [2, 1, 2, 3, 3], [10], [5, 5, 3]]
[12, 11, 10, 13]

Before spread with average of 14:
[[4, 3, 3, 2, 2, 2, 2, 1, 1], [10, 5, 3, 2, 1], [3, 3, 3], [6]]
Result and sum of each bucket:
[[3, 3, 3, 2, 3], [6, 1, 1, 2, 2, 1], [4, 3, 3, 2, 2], [10, 5]]
[14, 13, 14, 15]

Before spread with average of 4:
[[1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1], [1, 1]]
Result and sum of each bucket:
[[1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1, 1]]
[4, 4, 4, 4, 4]

Before spread with average of 14:
[[10, 9, 8, 7], [6, 5, 4], [3, 2], [1]]
Result and sum of each bucket:
[[1, 7, 9], [10], [6, 5, 4], [3, 2, 8]]
[17, 10, 15, 13]

Answer 3

这不是其他人建议的垃圾箱包装。那里的垃圾箱大小是固定的，您正在尝试最小化数量。在这里，您试图最小化固定数量的 bin 之间的差异。

事实证明这相当于Multiprocessor Scheduling，并且 - 根据参考资料 - 下面的算法（称为“最长作业优先”或“最长处理时间优先”）肯定会产生不超过 4/3 的最大总和 - 1/(3m) 次最优，其中 m 是桶数。在显示的测试用例中，我们将有 4/3-1/12 = 5/4 或不超过最佳值的 25%。

我们只是从空的所有垃圾箱开始，然后将每个项目按大小递减的顺序放入当前最不满的垃圾箱中。我们可以使用最小堆有效地跟踪最少满的 bin。对于具有 O(log n) insert 和 deletemin 的堆，该算法具有 O(n log m) 时间（n 和 m 定义为 @Jonas Elfström 所说）。Ruby 在这里非常有表现力：算法本身只有 9 个 sloc。

这是代码。我不是 Ruby 专家，所以请随时提出更好的方法。我正在使用@Jonas Elfström 的测试用例。

require 'algorithms'
require 'pp'

test_buckets = [ 
  [ [10, 4, 3, 3, 2, 1], [5, 5, 3, 2, 1], [3, 1, 1], [2] ],
  [ [4, 3, 3, 2, 2, 2, 2, 1, 1], [10, 5, 3, 2, 1], [3, 3, 3], [6] ],
  [ [1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1], [1, 1, 1, 1], [1, 1, 1], [1, 1] ],
  [ [10, 9, 8, 7], [6, 5, 4], [3, 2], [1] ],
]

def relevel(buckets)
  q = Containers::PriorityQueue.new { |x, y| x < y }

  # Initially all buckets to be returned are empty and so have zero sums.
  rtn = Array.new(buckets.length) { [] } 
  buckets.each_index {|i| q.push(i, 0) }
  sums = Array.new(buckets.length, 0)

  # Add to emptiest bucket in descending order. 
  # Bang! ops would generate less garbage.
  buckets.flatten.sort.reverse.each do |val|
    i = q.pop                 # Get index of emptiest bucket
    rtn[i] << val             # Append current value to it
    q.push(i, sums[i] += val) # Update sums and min heap
  end
  rtn
end

test_buckets.each {|b| pp relevel(b).map {|a| a.inject(:+) }}

结果：

[12, 11, 11, 12]
[14, 14, 14, 14]
[4, 4, 4, 4, 4]
[13, 13, 15, 14]

Answer 4

您可以使用我的答案将 n 可变高度图像拟合到 3（相似长度）列布局中。

心理地图：

• 对象大小到图片高度，以及
• 桶数到 bincount

然后该解决方案的其余部分应该适用......

---

下面使用了 Robin Green 之前提到的 first_fit 算法，但随后通过贪婪交换对此进行了改进。

交换例程找到离平均列高最远的列，然后系统地在它的一个图片和另一列中的第一张图片之间寻找一个交换，以最小化与平均值的最大偏差。

我使用了 30 张图片的随机样本，高度在 5 到 50 个“单位”之间。就我而言，收敛速度很快，并且在 first_fit 算法上得到了显着改进。

代码（Python 3.2：

def first_fit(items, bincount=3):
    items = sorted(items, reverse=1) # New - improves first fit.
    bins     = [[] for c in range(bincount)]
    binsizes = [0] * bincount
    for item in items:
        minbinindex = binsizes.index(min(binsizes))
        bins[minbinindex].append(item)
        binsizes[minbinindex] += item
    average = sum(binsizes) / float(bincount)
    maxdeviation = max(abs(average - bs) for bs in binsizes)

    return bins, binsizes, average, maxdeviation

def swap1(columns, colsize, average, margin=0):
    'See if you can do a swap to smooth the heights'
    colcount = len(columns)
    maxdeviation, i_a = max((abs(average - cs), i)
                              for i,cs in enumerate(colsize))
    col_a = columns[i_a]
    for pic_a in set(col_a): # use set as if same height then only do once
        for i_b, col_b in enumerate(columns):
            if i_a != i_b: # Not same column
                for pic_b in set(col_b):
                    if (abs(pic_a - pic_b) > margin): # Not same heights
                        # new heights if swapped
                        new_a = colsize[i_a] - pic_a + pic_b
                        new_b = colsize[i_b] - pic_b + pic_a
                        if all(abs(average - new) < maxdeviation
                               for new in (new_a, new_b)):
                            # Better to swap (in-place)
                            colsize[i_a] = new_a
                            colsize[i_b] = new_b
                            columns[i_a].remove(pic_a)
                            columns[i_a].append(pic_b)
                            columns[i_b].remove(pic_b)
                            columns[i_b].append(pic_a)
                            maxdeviation = max(abs(average - cs)
                                               for cs in colsize)
                            return True, maxdeviation
    return False, maxdeviation

def printit(columns, colsize, average, maxdeviation):
    print('columns')
    pp(columns)
    print('colsize:', colsize)
    print('average, maxdeviation:', average, maxdeviation)
    print('deviations:', [abs(average - cs) for cs in colsize])
    print()


if __name__ == '__main__':
    ## Some data
    #import random
    #heights = [random.randint(5, 50) for i in range(30)]
    ## Here's some from the above, but 'fixed'.
    from pprint import pprint as pp

    heights = [45, 7, 46, 34, 12, 12, 34, 19, 17, 41,
               28, 9, 37, 32, 30, 44, 17, 16, 44, 7,
               23, 30, 36, 5, 40, 20, 28, 42, 8, 38]

    columns, colsize, average, maxdeviation = first_fit(heights)
    printit(columns, colsize, average, maxdeviation)
    while 1:
        swapped, maxdeviation = swap1(columns, colsize, average, maxdeviation)
        printit(columns, colsize, average, maxdeviation)
        if not swapped:
            break
        #input('Paused: ')

输出：

columns
[[45, 12, 17, 28, 32, 17, 44, 5, 40, 8, 38],
 [7, 34, 12, 19, 41, 30, 16, 7, 23, 36, 42],
 [46, 34, 9, 37, 44, 30, 20, 28]]
colsize: [286, 267, 248]
average, maxdeviation: 267.0 19.0
deviations: [19.0, 0.0, 19.0]

columns
[[45, 12, 17, 28, 17, 44, 5, 40, 8, 38, 9],
 [7, 34, 12, 19, 41, 30, 16, 7, 23, 36, 42],
 [46, 34, 37, 44, 30, 20, 28, 32]]
colsize: [263, 267, 271]
average, maxdeviation: 267.0 4.0
deviations: [4.0, 0.0, 4.0]

columns
[[45, 12, 17, 17, 44, 5, 40, 8, 38, 9, 34],
 [7, 34, 12, 19, 41, 30, 16, 7, 23, 36, 42],
 [46, 37, 44, 30, 20, 28, 32, 28]]
colsize: [269, 267, 265]
average, maxdeviation: 267.0 2.0
deviations: [2.0, 0.0, 2.0]

columns
[[45, 12, 17, 17, 44, 5, 8, 38, 9, 34, 37],
 [7, 34, 12, 19, 41, 30, 16, 7, 23, 36, 42],
 [46, 44, 30, 20, 28, 32, 28, 40]]
colsize: [266, 267, 268]
average, maxdeviation: 267.0 1.0
deviations: [1.0, 0.0, 1.0]

columns
[[45, 12, 17, 17, 44, 5, 8, 38, 9, 34, 37],
 [7, 34, 12, 19, 41, 30, 16, 7, 23, 36, 42],
 [46, 44, 30, 20, 28, 32, 28, 40]]
colsize: [266, 267, 268]
average, maxdeviation: 267.0 1.0
deviations: [1.0, 0.0, 1.0]

好问题。

---

这是我在下面的单独评论中提到的反向排序信息。

>>> h = sorted(heights, reverse=1)
>>> h
[46, 45, 44, 44, 42, 41, 40, 38, 37, 36, 34, 34, 32, 30, 30, 28, 28, 23, 20, 19, 17, 17, 16, 12, 12, 9, 8, 7, 7, 5]
>>> columns, colsize, average, maxdeviation = first_fit(h)
>>> printit(columns, colsize, average, maxdeviation)
columns
[[46, 41, 40, 34, 30, 28, 19, 12, 12, 5],
 [45, 42, 38, 36, 30, 28, 17, 16, 8, 7],
 [44, 44, 37, 34, 32, 23, 20, 17, 9, 7]]
colsize: [267, 267, 267]
average, maxdeviation: 267.0 0.0
deviations: [0.0, 0.0, 0.0]

---

如果您有反向排序，附加到上述代码底部的这个额外代码（在 'if name == ... 中）将对随机数据进行额外试验：

for trial in range(2,11):
    print('\n## Trial %i' % trial)
    heights = [random.randint(5, 50) for i in range(random.randint(5, 50))]
    print('Pictures:',len(heights))
    columns, colsize, average, maxdeviation = first_fit(heights)
    print('average %7.3f' % average, '\nmaxdeviation:')
    print('%5.2f%% = %6.3f' % ((maxdeviation * 100. / average), maxdeviation))
    swapcount = 0
    while maxdeviation:
        swapped, maxdeviation = swap1(columns, colsize, average, maxdeviation)
        if not swapped:
            break
        print('%5.2f%% = %6.3f' % ((maxdeviation * 100. / average), maxdeviation))
        swapcount += 1
    print('swaps:', swapcount)

额外的输出显示了交换的效果：

## Trial 2
Pictures: 11
average  72.000 
maxdeviation:
 9.72% =  7.000
swaps: 0

## Trial 3
Pictures: 14
average 118.667 
maxdeviation:
 6.46% =  7.667
 4.78% =  5.667
 3.09% =  3.667
 0.56% =  0.667
swaps: 3

## Trial 4
Pictures: 46
average 470.333 
maxdeviation:
 0.57% =  2.667
 0.35% =  1.667
 0.14% =  0.667
swaps: 2

## Trial 5
Pictures: 40
average 388.667 
maxdeviation:
 0.43% =  1.667
 0.17% =  0.667
swaps: 1

## Trial 6
Pictures: 5
average  44.000 
maxdeviation:
 4.55% =  2.000
swaps: 0

## Trial 7
Pictures: 30
average 295.000 
maxdeviation:
 0.34% =  1.000
swaps: 0

## Trial 8
Pictures: 43
average 413.000 
maxdeviation:
 0.97% =  4.000
 0.73% =  3.000
 0.48% =  2.000
swaps: 2

## Trial 9
Pictures: 33
average 342.000 
maxdeviation:
 0.29% =  1.000
swaps: 0

## Trial 10
Pictures: 26
average 233.333 
maxdeviation:
 2.29% =  5.333
 1.86% =  4.333
 1.43% =  3.333
 1.00% =  2.333
 0.57% =  1.333
swaps: 4

Answer 5

调整背包问题求解算法，例如，将每个桶的“权重”指定为大致等于 n 个对象大小的平均值（尝试围绕平均值的高斯分布）。

http://en.wikipedia.org/wiki/Knapsack_problem#Solving

Answer 6

按大小顺序对存储桶进行排序。

将一个对象从最大的桶移动到最小的桶，对数组重新排序（这是几乎排序的，所以我们可以在两个方向上使用“有限插入排序”；您还可以通过注意放置最后一个对象的位置来加快速度两个要排序的桶。如果您有 6-6-6-6-6-6-5... 并从第一个桶中获取一个对象，则将其移动到第六个位置。然后在下一次迭代中，您可以从第五个开始比较。对于最小的桶，从右到左也是如此）。

当两桶相差一时，即可停止。

这移动了最小数量的桶，但为了比较，顺序为n^2 log n（最简单的版本是 n^3 log n）。如果对象移动很昂贵而桶大小检查不是，那么对于合理的 n 它可能仍然可以：

12 7 5 2
11 7 5 3
10 7 5 4
 9 7 5 5
 8 7 6 5
 7 7 6 6

12 7 3 1
11 7 3 2
10 7 3 3
 9 7 4 3
 8 7 4 4
 7 7 5 4
 7 6 5 5
 6 6 6 5

另一种可能性是计算每个桶的预期平均大小，并“移动”一个袋子（或另一个桶），将多余的部分从较大的桶转移到较小的桶。

否则，可能会发生奇怪的事情：

12 7 3 1, the average is a bit less than 6, so we take 5 as the average.

5 7 3 1  bag = 7 from 1st bucket
5 5 3 1  bag = 9
5 5 5 1  bag = 7
5 5 5 8  which is a bit unbalanced.

通过取 6（即四舍五入）它会更好，但有时它又不起作用：

12 5 3 1
 6 5 3 1  bag = 6 from 1st bucket
 6 6 3 1  bag = 5
 6 6 6 1  bag = 2
 6 6 6 3  which again is unbalanced.

你可以运行两次，第一次是从左到右的舍入平均值，另一个是从右到左截断的平均值：

12 5 3 1  we want to get no more than 6 in each bucket
6 11 3 1
6  6 8 1
 6 6 6 3
 6 6 6 3  and now we want to get at least 5 in each bucket
 6 6 4 5  (we have taken 2 from bucket #3 into bucket #5)
 6 5 5 5  (when the difference is 1 we stop).

这将需要“n log n”大小检查，并且不超过 2n 个对象移动。

另一种有趣的可能性是这样推理：您将 m 个对象放入 n 个桶中。所以你需要做一个m到n的整数映射，这就是 Bresenham 的线性化算法。在排序后的数组上运行 (n,m) Bresenham，在第 i 步（即针对第 i 个桶），算法将告诉您是使用 round(m/n) 还是 floor(m/n) 大小。然后根据第 i 个桶的大小将物体从“移动袋”移出或移入“移动袋”。

这需要n log n比较。

您可以通过最初将所有大小为 round(m/n) 或 floor(m/n) 的存储桶移除到两个大小为 R 或 F 的存储桶池来进一步减少对象移动的数量。运行算法时，您需要第 i 个桶来保存 R 个对象，如果 R 个对象的池不为空，则将第 i 个桶与 R 大小的桶中的一个交换。这样，只有大小过小或过大的桶才能平衡。（大多数）其他人被简单地忽略了，除了他们的引用被打乱了。

如果对象访问时间与计算时间成正比（例如某种自动加载弹匣），这将产生一个尽可能平衡的弹匣，整个对象移动的绝对最小值。

Answer 7

如果速度足够快，您可以使用整数编程包。

正确设置约束可能很棘手。像下面这样的东西可能会奏效：

让变量Oij表示Object i在Bucket j。让Wi代表重量或大小Oi

约束：

sum(Oij for all j) == 1       #each object is in only one bucket
Oij = 1 or 0.                 #object is either in bucket j or not in bucket j
sum(Oij * Wi for all i) <= X + R   #restrict weight on buckets.

客观的：

minimize X

请注意R，您可以根据需要多少运动和需要多少性能来调整松弛常数。

现在maximum bucket size是X + R

下一步是找出可能的最小移动量，同时保持桶大小小于X + R

定义一个 Stay 变量Si来控制是否Oi留在bucket j

如果Si是0，则表明它Oi停留在原来的位置。

约束：

Si = 1 or 0.
Oij = 1 or 0.
Oij <= Si where j != original bucket of Object i
Oij != Si where j == original bucket of Object i
Sum(Oij for all j) == 1
Sum(Oij for all i) <= X + R

客观的：

minimize Sum(Si for all i)

这里Sum(Si for all i)表示已移动的对象数。