let*_*aik 9 algorithm scala matrix scala-2.8
我的问题很简单,但我还没有找到有效的实现.
假设有一个像这样的矩阵A:
0 0 0 0 0 0 0
4 4 2 2 2 0 0
4 4 2 2 2 0 0
0 0 2 2 2 1 1
0 0 0 0 0 1 1
现在我想找到这个矩阵中具有给定大小的矩形区域的所有起始位置.区域是A的子集,其中所有数字都相同.
我们说宽度= 2和高度= 3.有3个区域有这样的大小:
2 2 2 2 0 0
2 2 2 2 0 0
2 2 2 2 0 0
函数调用的结果将是这些区域的起始位置(x,y从0开始)的列表.
List((2,1),(3,1),(5,0))
以下是我目前的实施情况."区域"在这里被称为"表面".
case class Dimension2D(width: Int, height: Int)
case class Position2D(x: Int, y: Int)
def findFlatSurfaces(matrix: Array[Array[Int]], surfaceSize: Dimension2D): List[Position2D] = {
val matrixWidth = matrix.length
val matrixHeight = matrix(0).length
var resultPositions: List[Position2D] = Nil
for (y <- 0 to matrixHeight - surfaceSize.height) {
var x = 0
while (x <= matrixWidth - surfaceSize.width) {
val topLeft = matrix(x)(y)
val topRight = matrix(x + surfaceSize.width - 1)(y)
val bottomLeft = matrix(x)(y + surfaceSize.height - 1)
val bottomRight = matrix(x + surfaceSize.width - 1)(y + surfaceSize.height - 1)
// investigate further if corners are equal
if (topLeft == bottomLeft && topLeft == topRight && topLeft == bottomRight) {
breakable {
for (sx <- x until x + surfaceSize.width;
sy <- y until y + surfaceSize.height) {
if (matrix(sx)(sy) != topLeft) {
x = if (x == sx) sx + 1 else sx
break
}
}
// found one!
resultPositions ::= Position2D(x, y)
x += 1
}
} else if (topRight != bottomRight) {
// can skip x a bit as there won't be a valid match in current row in this area
x += surfaceSize.width
} else {
x += 1
}
}
}
return resultPositions
}
我已经尝试在其中包含一些优化,但我确信有更好的解决方案.是否存在我可以移植的matlab函数？我也想知道这个问题是否有自己的名字,因为我不知道谷歌的用途.
谢谢你的思考!我很高兴看到你的建议或解决方案:)
编辑:我的应用程序中的矩阵尺寸约为300x300至3000x3000.此外,该算法仅针对相同的矩阵调用一次.原因是矩阵之后总是会改变(大约1-20%).
我实现了Kevin,Nikita和Daniel的算法,并在我的应用程序环境中对它们进行了基准测试,即这里没有孤立的综合基准测试,但是特别注意以最高性能的方式集成所有算法,这对Kevin的方法尤为重要,因为它使用了泛型(见下文).
首先,原始结果,使用Scala 2.8和jdk 1.6.0_23.作为解决特定于应用程序的问题的一部分,算法被执行了数百次."持续时间"表示应用程序算法完成之前所需的总时间(当然没有jvm启动等).我的机器是2.8GHz Core 2 Duo,有2个内核和2gig内存,-Xmx800M给了JVM.
重要说明:我认为我的基准设置对于像Daniel这样的并行算法来说并不公平.这是因为应用程序已经在计算多线程.所以这里的结果可能只显示相当于单线程速度.
矩阵大小233x587:
duration | JVM memory | avg CPU utilization
original O(n^4) | 3000s 30M 100%
original/-server| 840s 270M 100%
Nikita O(n^2) | 5-6s 34M 70-80%
Nikita/-server | 1-2s 300M 100%
Kevin/-server | 7400s 800M 96-98%
Kevin/-server** | 4900s 800M 96-99%
Daniel/-server | 240s 360M 96-99%
**使用@specialized,通过避免类型擦除来使泛型更快
矩阵大小2000x3000:
duration | JVM memory | avg CPU utilization
original O(n^4) | too long 100M 100%
Nikita O(n^2) | 150s 760M 70%
Nikita/-server | 295s (!) 780M 100%
Kevin/-server | too long, didn't try
首先,关于记忆的小记.-server JVM选项使用更多内存,具有更多优化的优势,并且通常可以更快地执行.从第二个表中可以看出,使用-server选项时,Nikita的算法速度较慢,这显然是因为达到了内存限制.我认为,即使对于小矩阵,这也会降低Kevin的算法,因为功能方法无论如何都在使用更多的内存.为了消除记忆因子我也有一个50×50矩阵试了一次,然后凯文的花5secs和尼基塔的0secs(当然,几乎为0).因此,无论如何它仍然较慢,而不仅仅是因为内存.
从数字中可以看出,我显然会使用Nikita的算法,因为它很快,在我的情况下这绝对是必要的.Daniel也指出,它也可以很容易地并行化.唯一的缺点是它不是真正的scala方式.
目前Kevin的算法通常有点过于复杂,因此很慢,但我确信可以进行更多优化(参见他的回答中的最后评论).
为了将Nikita的算法直接转换为功能风格,Daniel提出了一个已经非常快速的解决方案,如果他能使用scanRight,他说甚至会更快(参见他的回答中的最后评论).
下一步是什么？
在技术方面:等待Scala 2.9,ScalaCL,并进行综合基准测试以获得原始速度.
我所有这一切的目标都是拥有功能代码,但前提是它不会牺牲太多的速度.
选择答案:
至于选择答案,我想将尼基塔和丹尼尔的算法标记为答案,但我必须选择一个.我的问题标题包括"最有效",其中一个是命令式最快,另一个是功能式.虽然这个问题被标记为Scala我选择了尼基塔的命令式算法,因为2s对240s仍然是我接受的太多差异.我确定差异仍然可以推迟一点,任何想法？
所以,非常感谢你们!虽然我不会用功能的算法还,我得到了许多新的见解斯卡拉,我觉得我慢慢地得到所有的功能crazyness及其潜在的理解.(当然,即使没有进行太多的函数式编程,Scala也比Java更令人愉悦......这是学习它的另一个原因)
首先,一些辅助函数:
//count the number of elements matching the head
def runLength[T](xs:List[T]) = xs.takeWhile(_ == xs.head).size
//pair each element with the number of subsequent occurrences
def runLengths[T](row:List[T]) : List[(T,Int)] = row match {
case Nil => Nil
case h :: t => (h, runLength(row)) :: runLengths(t)
}
//should be optimised for tail-call, but easier to understand this way
//sample input: 1,1,2,2,2,3,4,4,4,4,5,5,6
//output: (1,2), (1,1), (2,3), (2,2), (2,1), (3,1), (4,4), (4,3), (4,2), (4,1), (5,2), (5,1), (6,1)
然后可以对网格中的每一行使用它:
val grid = List(
List(0,0,0,0),
List(0,1,1,0),
List(0,1,1,0),
List(0,0,0,0))
val stage1 = grid map runLengths
// returns stage1: List[List[(Int, Int)]] =
// 0,4 0,3 0,2 0,1
// 0,1 1,2 1,1 0,1
// 0,1 1,2 1,1 0,1
// 0,4 0,3 0,2 0,1
然后完成了水平行,我们现在对列执行完全相同的操作.这使用transposeScala标准集合库中可用的方法来交换行< - >列,这是根据同名的数学矩阵操作.一旦完成,我们也会转换回来.
val stage2 = (stage1.transpose map runLengths).transpose
// returns stage2: List[List[((Int, Int), Int)]] =
// (0,4),1 (0,3),1 (0,2),1 (0,1),4
// (0,1),2 (1,2),2 (1,1),2 (0,1),3
// (0,1),1 (1,2),1 (1,1),1 (0,1),2
// (0,4),1 (0,3),1 (0,2),1 (0,1),1
这是什么意思？取一个元素:(1,2),2,表示该单元格包含该值1,并向右扫描该行中包含2个相邻单元格1.向下扫描,有两个相邻的单元格具有相同的属性,包含该值1并且右侧具有相同数量的相等值.
整理后,将表单的嵌套元组转换((a,b),c)为(a,(b,c)):
val stage3 = stage2 map { _.map {case ((a,b),c) => a->(b,c) } }
//returns stage3: List[List[(Int, (Int, Int))]] =
// 0,(4,1) 0,(3,1) 0,(2,1) 0,(1,4)
// 0,(1,2) 1,(2,2) 1,(1,2) 0,(1,3)
// 0,(1,1) 1,(2,1) 1,(1,1) 0,(1,2)
// 0,(4,1) 0,(3,1) 0,(2,1) 0,(1,1)
其中1,(2,2)指的是包含该值的单元格1,并且位于具有相似值的单元格的2x2矩形的左上角.
从这里开始,发现相同大小的矩形是微不足道的.如果您还想要排除作为较大矩形子集的区域,则需要做更多工作.
更新:正如所写,该算法不适用于像(0,0)处的单元格那样的情况,它属于两个不同的矩形(1x4和4x1).如果需要,这也可以使用相同的技术解决.(使用map/transpose/map/transpose进行一次传递,使用transpose/map/transpose/map进行第二次传递,然后压缩并展平结果).
如果输入可能包含包含相同值的单元格的相邻矩形,则还需要修改,例如:
0 0 0 0 0 0 0 0
0 0 1 1 1 0 0 0
0 0 1 1 1 0 0 0
0 0 1 1 1 1 1 0
0 0 1 1 1 1 1 0
0 0 1 1 1 1 1 0
0 0 0 0 0 0 0 0
把它们放在一起,然后清理一下:
type Grid[T] = List[List[T]]
def runLengths[T](row:List[T]) : List[(T,Int)] = row match {
case Nil => Nil
case h :: t => (h -> row.takeWhile(_ == h).size) :: runLengths(t)
}
def findRectangles[T](grid: Grid[T]) = {
val step1 = (grid map runLengths)
val step2 = (step1.transpose map runLengths).transpose
step2 map { _ map { case ((a,b),c) => (a,(b,c)) } }
}
抓住你的帽子,这是一个很大的...
在编写一行新功能之前,我们首先重构一下,将一些方法拉入带有隐式转换的Ops类中,这样它们就可以像在底层集合类型上定义的方法一样使用:
import annotation.tailrec
class RowOps[T](row: List[T]) {
def withRunLengths[U](func: (T,Int)=>U) : List[U] = {
@tailrec def recurse(row:List[T], acc:List[U]): List[U] = row match {
case Nil => acc
case head :: tail =>
recurse(
tail,
func(head, row.takeWhile(head==).size) :: acc)
}
recurse(row, Nil).reverse
}
def mapRange(start: Int, len: Int)(func: T=>T) =
row.splitAt(start) match {
case (l,r) => l ::: r.take(len).map(func) ::: r.drop(len)
}
}
implicit def rowToOps[T](row: List[T]) = new RowOps(row)
这withRunLengths为列表添加了一种方法.这里一个值得注意的区别是,(value, length)该方法不是返回一对列表,而是接受一个函数来为每个这样的对创建一个输出值.这将在以后派上用场......
type Grid[T] = List[List[T]]
class GridOps[T](grid: Grid[T]) {
def deepZip[U](other: Grid[U]) = (grid zip other) map { case (g,o) => g zip o}
def deepMap[U](f: (T)=>U) = grid map { _ map f}
def mapCols[U](f: List[T]=>List[U]) = (grid.transpose map f).transpose
def height = grid.size
def width = grid.head.size
def coords = List.tabulate(height,width){ case (y,x) => (x,y) }
def zipWithCoords = deepZip(coords)
def deepMapRange(x: Int, y: Int, w: Int, h: Int)(func: T=>T) =
grid mapRange (y,h){ _.mapRange(x,w)(func) }
}
implicit def gridToOps[T](grid: Grid[T]) = new GridOps(grid)
这里不应该有任何意外.这些deepXXX方法避免了编写表单的构造list map { _ map { ... } }. tabulate对你来说也可能是新的,但它的目的是希望从使用中显而易见.
使用这些,定义用于在整个网格上查找水平和垂直游程长度的函数变得微不足道:
def withRowRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U) =
grid map { _.withRunLengths(func) }
def withColRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U) =
grid mapCols { _.withRunLengths(func) }
为什么2个参数块而不是一个？我马上解释一下.
我可以将这些定义为方法GridOps,但它们似乎不适合一般用途.随意在这里不同意我:)
接下来,定义一些输入......
def parseIntGrid(rows: String*): Grid[Int] =
rows.toList map { _ map {_.toString.toInt} toList }
val input: Grid[Int] = parseIntGrid("0000","0110","0110","0000")
......另一个有用的助手类型......
case class Rect(w: Int, h: Int)
object Rect { def empty = Rect(0,0) }
作为元组的替代,这确实有助于调试.深度嵌套的括号在眼睛上并不容易.(对不起Lisp粉丝!)
...并使用上面定义的函数:
val stage1w = withRowRunLengths(input) {
case (cell,width) => (cell,width)
}
val stage2w = withColRunLengths(stage1w) {
case ((cell,width),height) => Rect(width,height)
}
val stage1t = withColRunLengths(input) {
case (cell,height) => (cell,height)
}
val stage2t = withRowRunLengths(stage1t) {
case ((cell,height),width) => Rect(width,height)
}
以上所有块都应该是单行,但我重新格式化了StackOverflow.
这个阶段的输出只是矩形的网格,我故意放弃了矩形所包含的实际值的任何提及.这绝对没问题,从网格中的坐标中找到它很容易,我们将在短时间内重新组合数据.
还记得我解释一下RowOps#withRunLengths将函数作为参数吗？嗯,这是它被使用的地方. case (cell,width) => (cell,width)实际上是取单元值和游程长度(调用它们的功能cell和width),然后返回该(cell,width)元组.
这也是我在定义函数时使用两个参数块的原因,因此第二个参数可以在{braces}中传递,并使整个事物都很好并且类似于DSL.
这里说明的另一个非常重要的原则是类型推理器连续处理参数块,所以为此(记得吗？):
def withRowRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U)
类型T将由提供的网格确定.编译器然后知道作为第二个参数提供的函数的输入类型,Int在这种情况下,因为第一个参数是Grid[Int] - 这就是为什么我能够写入case (cell,width) => (cell,width)而不必显式地声明任何地方cell和width整数.在第二种用法中,Grid[(Int,Int)]提供了a ,这适合封闭,case ((cell,width),height) => Rect(width,height)并且它再次起作用.
如果该闭包包含任何不适用于网格底层类型的东西,那么编译器会抱怨,这就是类型安全的全部内容......
计算完所有可能的矩形之后,剩下的就是收集数据并以更实用的格式呈现数据.因为在这个阶段的嵌套可能会非常混乱,我定义了另一种数据类型:
case class Cell[T](
value: T,
coords: (Int,Int) = (0,0),
widest: Rect = Rect.empty,
tallest: Rect = Rect.empty
)
这里没什么特别的,只是一个带有命名/默认参数的case类.我也很高兴我有远见来定义Rect.empty上面:)
现在混合4个数据集(输入vals,coords,最宽的rects,最高的rects),逐渐折叠到细胞中,轻轻搅拌,并服务:
val cellsWithCoords = input.zipWithCoords deepMap {
case (v,(x,y)) => Cell(value=v, coords=(x,y))
}
val cellsWithWidest = cellsWithCoords deepZip stage2w deepMap {
case (cell,rect) => cell.copy(widest=rect)
}
val cellsWithWidestAndTallest = cellsWithWidest deepZip stage2t deepMap {
case (cell,rect) => cell.copy(tallest=rect)
}
val results = (cellsWithWidestAndTallest deepMap {
case Cell(value, coords, widest, tallest) =>
List((widest, value, coords), (tallest, value, coords))
}
).flatten.flatten
最后一个阶段很有意思,它将每个单元格转换为(矩形,值,坐标)元组的大小为2的列表(大小为2,因为我们对每个单元格都有最宽和最高的区域).调用两次展平然后将结果List[List[List[_]]]返回到单个List.由于必要的坐标已嵌入结果中,因此无需再保留2D结构.
瞧!
现在由你来决定你现在用这个List做什么.下一阶段可能是某种形式的排序和重复删除......
你可以O(n^2)相对容易地做到这一点.
首先,一些预先计算.对于矩阵中的每个单元格,计算它下面有多少个连续单元格具有相同的数字.
对于您的示例,结果矩阵a(想不出更好的名称:/)将如下所示
0 0 0 0 0 2 2
1 1 2 2 2 1 1
0 0 1 1 1 0 0
1 1 0 0 0 1 1
0 0 0 0 0 0 0
它可以O(n^2)很容易地生产.
现在,对于每一行i,让我们找到所有行的顶边i(和行的底边i + height - 1)的矩形.
这是一个插图i = 1
0 0 0 0 0 0 0
-------------
4 4 2 2 2 0 0
4 4 2 2 2 0 0
0 0 2 2 2 1 1
-------------
0 0 0 0 0 1 1
现在,主要的想法
int current_width = 0;
for (int j = 0; j < matrix.width; ++j) {
if (a[i][j] < height - 1) {
// this column has different numbers in it, no game
current_width = 0;
continue;
}
if (current_width > 0) {
// this column should consist of the same numbers as the one before
if (matrix[i][j] != matrix[i][j - 1]) {
current_width = 1; // start streak anew, from the current column
continue;
}
}
++current_width;
if (current_width >= width) {
// we've found a rectangle!
}
}
在上面的例子(i = 1)current_width之后每次迭代都会0, 0, 1, 2, 3, 0, 0.
现在,我们需要迭代所有可能的i,我们有一个解决方案.
我会在这里扮演魔鬼的拥护者.我将展示以功能样式编写的Nikita精确算法.我也将它并行化,只是为了表明它可以完成.
首先,计算单元格下方具有相同编号的连续单元格.我通过返回所有值加上一个与Nikita的建议输出相比稍微改变,以避免- 1代码的其他部分.
def computeHeights(column: Array[Int]) = (
column
.reverse
.sliding(2)
.map(pair => pair(0) == pair(1))
.foldLeft(List(1)) (
(list, flag) => (if (flag) list.head + 1 else 1) :: list
)
)
我宁愿.view在倒车前使用,但这不适用于目前的Scala版本.如果它这样做,它将节省重复的阵列创建,这应该加速代码很多,因为内存位置和带宽原因,如果没有其他.
现在,所有列同时:
import scala.actors.Futures.future
def getGridHeights(grid: Array[Array[Int]]) = (
grid
.transpose
.map(column => future(computeHeights(column)))
.map(_())
.toList
.transpose
)
我不确定并行化开销是否会在这里得到回报,但这是Stack Overflow上的第一个算法,它实际上有机会,考虑到计算列的非常重要的工作.这是使用即将推出的2.9功能编写它的另一种方式(它可能适用于Scala 2.8.1 - 不确定:
def getGridHeights(grid: Array[Array[Int]]) = (
grid
.transpose
.toParSeq
.map(computeHeights)
.toList
.transpose
)
现在,尼基塔算法的肉:
def computeWidths(height: Int, row: Array[Int], heightRow: List[Int]) = (
row
.sliding(2)
.zip(heightRow.iterator)
.toSeq
.reverse
.foldLeft(List(1)) { case (widths @ (currWidth :: _), (Array(prev, curr), currHeight)) =>
(
if (currHeight >= height && currWidth > 0 && prev == curr) currWidth + 1
else 1
) :: widths
}
.toArray
)
我在这段代码中使用模式匹配,虽然我关心它的速度,因为所有的滑动,拉链和折叠都有两个很多东西在这里玩杂耍.而且,谈到性能,我正在使用Array而不是IndexedSeq因为它Array是JVM中唯一没有被擦除的类型,因此可以获得更好的性能Int.然后,.toSeq由于内存局部性和带宽,我也不是特别高兴.
另外,我是从右到左而不是Nikita从左到右做的,所以我可以找到左上角.
然而,与Nikita的答案中的代码是一样的,除了我仍然在他的代码中添加一个当前宽度的事实,而不是在这里打印结果.有没有在这里明确的起源了一堆东西,虽然- ,,row ...让我们来看看在上下文中的代码-和并行! - 了解整体情况.heightRowheight
def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
grid
.zip(getGridHeights(grid))
.map { case (row, heightsRow) => future(computeWidths(height, row, heightsRow)) }
.map(_())
)
和2.9版本:
def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
grid
.toParSeq
.zip(getGridHeights(grid))
.map { case (row, heightsRow) => computeWidths(height, row, heightsRow) }
)
并且,对于结局,
def findRectangles(height: Int, width: Int, grid: Array[Array[Int]]) = {
val gridWidths = getGridWidths(height, grid)
for {
y <- gridWidths.indices
x <- gridWidths(y).indices
if gridWidths(y)(x) >= width
} yield (x, y)
}
所以...我毫不怀疑Nikita算法的命令式版本更快 - 它只使用Array,原始版本比任何其他类型快得多,并且它避免了大量创建临时集合 - 尽管Scala本可以做得更好这里.此外,没有倒闭-尽可能多的帮助,他们是比代码慢而不关闭.至少在JVM成长起来帮助他们之前.
此外,尼基塔的代码可以很容易地与线程并行化 - 所有东西! - 没什么问题.
但我的观点是,尼基塔的代码并不是特别糟糕,因为它在这里和那里使用数组和一个可变变量.该算法干净利落地转换为更实用的风格.
编辑
所以,我决定尝试制作一个高效的功能版本.它并没有真正完全正常运行,因为我使用它Iterator,这是可变的,但它足够接近.不幸的是,它不会在斯卡拉2.8.1工作,因为它缺乏scanLeft上Iterator.
这里还有另外两件不幸的事情.首先,我放弃了网格高度的并行化,因为我无法绕过至少有一个transpose,所有的集合复制都需要.尽管如此,仍然至少有一次数据复制(参见toArray了解其中的地方).
此外,由于我正在使用,Iterable我失去了使用并行集合的能力.我确实想知道grid从一开始就是并行集合的并行集合是否会使代码变得更好.
我不知道这是否比以前的版本更有效.这是一个有趣的问题......
def getGridHeights(grid: Array[Array[Int]]) = (
grid
.sliding(2)
.scanLeft(Array.fill(grid.head.size)(1)) { case (currHeightArray, Array(prevRow, nextRow)) =>
(prevRow, nextRow, currHeightArray)
.zipped
.map { case (x, y, currHeight) => if (x == y) currHeight + 1 else 1 }
}
)
def computeWidths(height: Int, row: Array[Int], heightRow: Array[Int]) = (
row
.sliding(2)
.map { case Array(x, y) => x == y }
.zip(heightRow.iterator)
.scanLeft(1) { case (currWidth , (isConsecutive, currHeight)) =>
if (currHeight >= height && currWidth > 0 && isConsecutive) currWidth + 1
else 1
}
.toArray
)
import scala.actors.Futures.future
def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
grid
.iterator
.zip(getGridHeights(grid))
.map { case (row, heightsRow) => future(computeWidths(height, row, heightsRow)) }
.map(_())
.toArray
)
def findRectangles(height: Int, width: Int, grid: Array[Array[Int]]) = {
val gridWidths = getGridWidths(height, grid)
for {
y <- gridWidths.indices
x <- gridWidths(y).indices
if gridWidths(y)(x) >= width
} yield (x - width + 1, y - height + 1)
}