函数式编程之pipeline——很酷有没有

Pipelinepipeline 管道借鉴于Unix Shell的管道操作——把若干个命令串起来，前面命令的输出成为后面命令的输入，如此完成一个流式计算。（注：管道绝对是一个伟大的发明，他的设哲学就是KISS – 让每个功能就做一件事，并把这件事做到极致，软件或程序的拼装会变得更为简单和直观。这个设计理念影响非常深远，包括今天的Web Service，云计算，以及大数据的流式计算等等...

djph26741

1042人浏览 · 2019-07-24 09:41:00

djph26741 · 2019-07-24 09:41:00 发布

Pipeline

pipeline 管道借鉴于Unix Shell的管道操作——把若干个命令串起来，前面命令的输出成为后面命令的输入，如此完成一个流式计算。（注：管道绝对是一个伟大的发明，他的设哲学就是KISS – 让每个功能就做一件事，并把这件事做到极致，软件或程序的拼装会变得更为简单和直观。这个设计理念影响非常深远，包括今天的Web Service，云计算，以及大数据的流式计算等等）

比如，我们如下的shell命令：

 
    
         ps  
         auwwx |  
         awk  
         '{print $2}'  
         |  
         sort  
         -n |  
         xargs  
         echo 
        
 
  

如果我们抽象成函数式的语言，就像下面这样：

 
         xargs(  echo, sort(n, awk( 
         'print $2' 
         , ps(auwwx)))  )

也可以类似下面这个样子：

 
         pids  
         =  
         for_each(result, [ps_auwwx, awk_p2, sort_n, xargs_echo])

好了，让我们来看看函数式编程的Pipeline怎么玩？

我们先来看一个如下的程序，这个程序的process()有三个步骤：

1）找出偶数。
2）乘以3
3）转成字符串返回

 
         def  
         process(num): 
        
         # filter out non-evens 
        
         if  
         num  
         %  
         2  
         ! 
         =  
         0 
         : 
        
         return 
        
         num  
         =  
         num  
         *  
         3 
        
         num  
         =  
         'The Number: %s'  
         %  
         num 
        
         return  
         num 
        
         nums  
         =  
         [ 
         1 
         ,  
         2 
         ,  
         3 
         ,  
         4 
         ,  
         5 
         ,  
         6 
         ,  
         7 
         ,  
         8 
         ,  
         9 
         ,  
         10 
         ] 
        
         for  
         num  
         in  
         nums: 
        
         print  
         process(num) 
        
         # 输出： 
        
         # None 
        
         # The Number: 6 
        
         # None 
        
         # The Number: 12 
        
         # None 
        
         # The Number: 18 
        
         # None 
        
         # The Number: 24 
        
         # None 
        
         # The Number: 30

我们可以看到，输出的并不够完美，另外，代码阅读上如果没有注释，你也会比较晕。下面，我们来看看函数式的pipeline（第一种方式）应该怎么写？

 
         def  
         even_filter(nums): 
        
         for  
         num  
         in  
         nums: 
        
         if  
         num  
         %  
         2  
         = 
         =  
         0 
         : 
        
         yield  
         num 
        
         def  
         multiply_by_three(nums): 
        
         for  
         num  
         in  
         nums: 
        
         yield  
         num  
         *  
         3 
        
         def  
         convert_to_string(nums): 
        
         for  
         num  
         in  
         nums: 
        
         yield  
         'The Number: %s'  
         %  
         num 
        
         nums  
         =  
         [ 
         1 
         ,  
         2 
         ,  
         3 
         ,  
         4 
         ,  
         5 
         ,  
         6 
         ,  
         7 
         ,  
         8 
         ,  
         9 
         ,  
         10 
         ] 
        
         pipeline  
         =  
         convert_to_string(multiply_by_three(even_filter(nums))) 
        
         for  
         num  
         in  
         pipeline: 
        
         print  
         num 
        
         # 输出： 
        
         # The Number: 6 
        
         # The Number: 12 
        
         # The Number: 18 
        
         # The Number: 24 
        
         # The Number: 30

我们动用了Python的关键字 yield，这个关键字主要是返回一个Generator，yield 是一个类似 return 的关键字，只是这个函数返回的是个Generator-生成器。所谓生成器的意思是，yield返回的是一个可迭代的对象，并没有真正的执行函数。也就是说，只有其返回的迭代对象被真正迭代时，yield函数才会正真的运行，运行到yield语句时就会停住，然后等下一次的迭代。（这个是个比较诡异的关键字）这就是lazy evluation。

好了，根据前面的原则——“使用Map & Reduce，不要使用循环”，那我们用比较纯朴的Map & Reduce吧。

 
         def  
         even_filter(nums): 
        
         return  
         filter 
         ( 
         lambda  
         x: x 
         % 
         2 
         = 
         = 
         0 
         , nums) 
        
         def  
         multiply_by_three(nums): 
        
         return  
         map 
         ( 
         lambda  
         x: x 
         * 
         3 
         , nums) 
        
         def  
         convert_to_string(nums): 
        
         return  
         map 
         ( 
         lambda  
         x:  
         'The Number: %s'  
         %  
         x,  nums) 
        
         nums  
         =  
         [ 
         1 
         ,  
         2 
         ,  
         3 
         ,  
         4 
         ,  
         5 
         ,  
         6 
         ,  
         7 
         ,  
         8 
         ,  
         9 
         ,  
         10 
         ] 
        
         pipeline  
         =  
         convert_to_string( 
        
         multiply_by_three( 
        
         even_filter(nums) 
        
         ) 
        
         ) 
        
         for  
         num  
         in  
         pipeline: 
        
         print  
         num

但是他们的代码需要嵌套使用函数，这个有点不爽，如果我们能像下面这个样子就好了（第二种方式）。

 
         pipeline_func(nums, [even_filter, 
        
         multiply_by_three, 
        
         convert_to_string])

那么，pipeline_func 实现如下：

 
         def  
         pipeline_func(data, fns): 
        
         return  
         reduce 
         ( 
         lambda  
         a, x: x(a), 
        
         fns, 
        
         data)

好了，在读过这么多的程序后，你可以回头看一下这篇文章的开头对函数式编程的描述，可能你就更有感觉了。

最后，我希望这篇浅显易懂的文章能让你感受到函数式编程的思想，就像OO编程，泛型编程，过程式编程一样，我们不用太纠结是不是我们的程序就是OO，就是functional的，我们重要的品味其中的味道。

参考

补充：评论中redraiment的这个评论大家也可以读一读。

感谢谢网友S142857 提供的shell风格的python pipeline：

 
         class  
         Pipe( 
         object 
         ): 
        
         def  
         __init__( 
         self 
         , func): 
        
         self 
         .func  
         =  
         func 
        
         def  
         __ror__( 
         self 
         , other): 
        
         def  
         generator(): 
        
         for  
         obj  
         in  
         other: 
        
         if  
         obj  
         is  
         not  
         None 
         : 
        
         yield  
         self 
         .func(obj) 
        
         return  
         generator() 
        
         @Pipe 
        
         def  
         even_filter(num): 
        
         return  
         num  
         if  
         num  
         %  
         2  
         = 
         =  
         0  
         else  
         None 
        
         @Pipe 
        
         def  
         multiply_by_three(num): 
        
         return  
         num 
         * 
         3 
        
         @Pipe 
        
         def  
         convert_to_string(num): 
        
         return  
         'The Number: %s'  
         %  
         num 
        
         @Pipe 
        
         def  
         echo(item): 
        
         print  
         item 
        
         return  
         item 
        
         def  
         force(sqs): 
        
         for  
         item  
         in  
         sqs:  
         pass 
        
         nums  
         =  
         [ 
         1 
         ,  
         2 
         ,  
         3 
         ,  
         4 
         ,  
         5 
         ,  
         6 
         ,  
         7 
         ,  
         8 
         ,  
         9 
         ,  
         10 
         ] 
        
         force(nums | even_filter | multiply_by_three | convert_to_string | echo)