泛函编程（13）－无穷数据流－Infinite Stream

Posted 2020-09-18 张宇航

    上节我们提到Stream和List的主要分别是在于Stream的“延后计算“（lazy evaluation）特性。我们还讨论过在处理大规模排列数据集时，Stream可以一个一个把数据元素搬进内存并且可以逐个元素地进行处理操作。这让我不禁联想到我们常用的数据搜索读取方式了：大量的数据存放在数据库里，就好像无穷的数据源头。我们把数据读取方式（那些数据库读写API函数）嵌入Stream的操作函数内，把数据搜索条件传入Stream构造器（constructor）中形成一个对数据搜索操作的描述。这个产生的Stream只有在我们调用符合搜索条件的数据库记录时才会逐个从数据库中读取出来。这可是一个非常熟悉的场景，但我们常常会思考它的原理。

    很多时我们都需要无穷数据流（infinite stream），以直接或一些算法方式有规则地重复产生数据。无穷数据流被定义为“反递归”（corecursive）的：递归的特性是从复杂到简单并最后终止，而无穷数据流的特性却是从简单到复杂并永远不会终结。

我们先从简单的无穷数据流开始：数字产生器：

1 def ones: Stream[Int] = cons(1,ones)              //> ones: => ch5.genstream.Stream[Int]
2 ones.take(5).toList                               //> res0: List[Int] = List(1, 1, 1, 1, 1)

 

ones函数可以产生一个无穷的数据流。每个元素都是常数1。从这个简单的例子我们可以稍微领略反递归的味道：cons(1,ones),通过重复不断运算cons来产生无穷数据。

我们可以把一些算法嵌入无穷数据流产生过程：

 

 1 def constant[A](a: A): Stream[A] = cons(a, constant(a))
 2                                                   //> constant: [A](a: A)ch5.genstream.Stream[A]
 3 constant(5).take(5).toList                        //> res1: List[Int] = List(5, 5, 5, 5, 5)
 4 
 5 def from(n: Int): Stream[Int] = cons(n, from(n+1))//> from: (n: Int)ch5.genstream.Stream[Int]
 6 from(5).take(5).toList                            //> res2: List[Int] = List(5, 6, 7, 8, 9)
 7 
 8 def fibs: Stream[Int] = {
 9     def go (prev: Int, cur: Int): Stream[Int] = {
10         cons(prev,go(cur,prev + cur))
11     }
12     go(0,1)
13 }                                                 //> fibs: => ch5.genstream.Stream[Int]
14 fibs.take(5).toList                               //> res3: List[Int] = List(0, 1, 1, 2, 3)

 

从以上这些例子可以看出：我们不断重复的在cons。而cons的参数则是算法的实现结果。

以下的unfold是一个最通用的Stream构建函数（stream builder），我们需要做些重点介绍：

1 def unfold[A,S](z: S)(f: S => Option[(A, S)]): Stream[A] = {
2     f(z) match {
3         case None => empty
4         case Some((a,s)) => cons(a, unfold(s)(f))
5     }
6 }                                                 //> unfold: [A, S](z: S)(f: S => Option[(A, S)])ch5.genstream.Stream[A]

unfold的工作原理模仿了一种状态流转过程：z是一个起始状态，代表的是一个类型的值。然后用户（caller）再提供一个操作函数f。f的款式是：S => Option[(A,S)]，意思是接受一个状态，然后把它转换成一对新的A值和新的状态S，再把它们放入一个Option。如果Option是None的话，这给了用户一个机会去终止运算，让unfold停止递归。从unfold的源代码可以看到f(z) match {} 的两种情况。需要注意的是函数f是针对z这个类型S来操作的，A类型是Stream［A]的元素类型。f的重点作用在于把S转换成新的S。我们用一些例子来说明：

1 def constantByUnfold[A](a: A): Stream[A] = unfold(a)(_ => Some((a,a)))
2                                                   //> constantByUnfold: [A](a: A)ch5.genstream.Stream[A]
3 constantByUnfold(2).take(5).toList                //> res4: List[Int] = List(2, 2, 2, 2, 2)

constantByUnfold产生一个无穷的常数：a同时代表了元素类型和状态。_ => Some((a,a))意思是无论输入任何状态，元素值和状态都不转变，所以unfold会产生同一个数字。另外f的结果永远不可能是None，所以这是一个无穷数据流（infinite stream）。

再来一个例子：

1 def fromByUnfold(s: Int): Stream[Int] = unfold(s)(s => Some(s,s+1))
2                                                   //> fromByUnfold: (s: Int)ch5.genstream.Stream[Int]
3 fromByUnfold(5).take(5).toList                    //> res5: List[Int] = List(5, 6, 7, 8, 9)

fromByUnfold产生一个从s开始的无穷整数序列：s 同时代表了元素类型和状态。_ => Some((s,s+1))表示新A值是s,新状态是s+1，所以新s = s + 1。状态转变原理可以从改变 s+1 到 s+2 运算后产生的结果得以了解：

 

1 def fromByUnfold_2(s: Int): Stream[Int] = unfold(s)(s => Some(s,s+2))
2                                                   //> fromByUnfold_2: (s: Int)ch5.genstream.Stream[Int]
3 fromByUnfold_2(5).take(5).toList                  //> res6: List[Int] = List(5, 7, 9, 11, 13)

再试一个不同类型的状态例子：

1 def fibsByUnfold: Stream[Int] = unfold((0,1)){ case (a1,a2) => Some((a1, (a2, a1+a2))) }
2                                                   //> fibsByUnfold: => ch5.genstream.Stream[Int]
3 fibsByUnfold.take(10).toList                      //> res8: List[Int] = List(0, 1, 1, 2, 3, 5, 8, 13, 21, 34)

在以上的例子里：S类型为tuple，起始值(0,1)，元素类型A是Int。函数f: Int => Option[(Int, Int)]。f函数返回新A=a1, 新状态S (a2, a1+a2)。由于状态是个tuple类型，(a1,a2)是个模式匹配操作，所以必须加上case。S=(0,1) >>> (A,S)=(0,(1,0+1)) >>>(1,(1,1+1))>>>(1,(2,2+1))>>>(2,(3,2+3))>>>(3,(5,3+5))>>>(5,(8,5+8))>>>(8,(13,8+13))从以上推断我们可以得出A>>>0,1,1,2,3,5,8,13，而状态S>>>(0,1),(1,1),(1,2),(2,3),(3,5),(5,8)...不断变化。?

在来个更体现状态转变的例子：用unfold实现的map函数 

1     def mapByUnfoldInfinite[B](f: A => B): Stream[B] = {
2             unfold(uncons) {
3             case Some((h,t)) => Some((f(h),Some((t.head, t.tail))))
4             case _ => None
5         }
6     }
7 (fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).take(5).toList
8                                                   //> res9: List[Int] = List(11, 12, 13, 14, 15)

S类型即uncons类型>>>Option[(A, Stream[A])], uncons的新状态是 Some((t.head, t.tail))。因为我们采用了数据结构嵌入式的设计，所以必须用uncons来代表Stream，它的下一个状态就是Some((t.head, t.tail))。如果使用子类方式Cons(h,t)，那么下一个状态就可以直接用t来表示，简洁多了。

 再看看还有什么函数可以用unfold来实现吧：

 

 1     def takeByUnfold(n: Int): Stream[A] = {
 2         unfold((uncons,n)) {
 3             case (Some((h,t)),k) if (k > 0) => Some(h, (Some((t.head, t.tail)), k-1))
 4             case _ => None
 5         }
 6     }
 7     def takeWhileByUnfold(f: A => Boolean): Stream[A] = {
 8         unfold(uncons) {
 9             case Some((h,t)) if (f(h)) => Some(h, Some((t.head, t.tail)))
10             case _ => None
11         }
12     }
13     def filterByUnfold(f: A => Boolean): Stream[A] = {
14         unfold(uncons) {
15             case Some((h,t)) if (f(h)) => Some(h, Some((t.head, t.tail)))
16             case _ => None
17         }
18     }
19         def zipWithByUnfold[B,C](b: Stream[B])(f: (A,B) => C): Stream[C] = {
20             unfold((uncons,b.uncons)) {
21                 case (Some((ha,ta)),Some((hb,tb))) => Some(f(ha,hb),(Some((ta.head,ta.tail)),Some((tb.head,tb.tail))))
22                 case _ => None
23             }
24         }
25     def zip[B](b: Stream[B]): Stream[(A,B)] = zipWithByUnfold(b){( _ , _)}
26 (fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).take(5).toList
27                                                   //> res9: List[Int] = List(11, 12, 13, 14, 15)
28 (fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).takeByUnfold(5).toList
29                                                   //> res10: List[Int] = List(11, 12, 13, 14, 15)
30 (fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).takeWhileByUnfold(_ < 15).toList
31                                                   //> res11: List[Int] = List(11, 12, 13, 14)
32 (fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).filterByUnfold(_ < 15).toList
33                                                   //> res12: List[Int] = List(11, 12, 13, 14)
34 (fromByUnfold(5) zip fromByUnfold(1).mapByUnfoldInfinite {_ + 10}).take(5).toList
35                                                   //> res13: List[(Int, Int)] = List((5,11), (6,12), (7,13), (8,14), (9,15))

 

风格基本上是一致的。

写完这些例子才有时间静下来想了一下：费了这么大劲搞这个无穷数据流到底能干什么呢？作为数据库搜索的数据源吗，这个可以用普通的Stream来实现。由于无穷数据流是根据一些算法有规则的不停顿产生数据，那么用来搭建测试数据源或者什么数学统计模式环境道是是可以的。想到不断产生数据，那么用来画些动态的东西也行吧，那在游戏软件中使用也有可能了。