怎么还蹦出来个 “ 数据管道 ”

简介: 怎么还蹦出来个 “ 数据管道 ”

创建数据处理管道


问题核心:★★   口感:天津肉包子


问题


你想以数据管道 (类似 Unix 管道) 的方式迭代处理数据。比如,你有个大量的数据  需要处理,但是不能将它们一次性放入内存中。


解决方案


生成器函数是一个实现管道机制的好办法。

假定你要处理一个非常大的  日志文件目录:


foo/
access-log-012007.gz
access-log-022007.gz
access-log-032007.gz
...
access-log-012008
bar/
access-log-092007.bz2
...
access-log-022008


假设每个日志文件包含这样的数据:

    124.115.6.12 - - [10/Jul/2012:00:18:50 -0500] "GET /robots.txt ..." 200 71
    210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] "GET /ply/ ..." 200 11875
    210.212.209.67 - - [10/Jul/2012:00:18:51 -0500] "GET /favicon.ico ..." 404 369
    61.135.216.105 - - [10/Jul/2012:00:20:04 -0500] "GET /blog/atom.xml ..." 304 -
    ...
    


    为了处理这些文件,你可以定义一个由多个执行特定任务独立任务的简单生成器函数组成的容器

    就像这样:

      import os
      import fnmatch
      import gzip
      import bz2
      import re
      def gen_find(filepat, top):
            '''
            查找目录树中与外壳通配符模式匹配的所有文件名
            '''
            for path, dirlist, filelist in os.walk(top):
                for name in fnmatch.filter(filelist, filepat):
                    yield os.path.join(path,name)
      def gen_opener(filenames):
            '''
            每次打开一个文件名序列,生成一个file对象。在进行下一次迭代时,文件将立即关闭。
            '''
            for filename in filenames:
                if filename.endswith('.gz'):
                    f = gzip.open(filename, 'rt')
                elif filename.endswith('.bz2'):
                    f = bz2.open(filename, 'rt')
                else:
                    f = open(filename, 'rt')
                yield f
                f.close()
      def gen_concatenate(iterators):
          '''
          将一个迭代器序列链接到一个单独的序列中
          '''
          for it in iterators:
              yield from it
      def gen_grep(pattern, lines):
          '''
          在一个行序列中查找regex模式
          '''
          pat = re.compile(pattern)
              for line in lines:
                  if pat.search(line):
                        yield line


      现在你可以很容易的将这些函数连起来创建一个处理管道。比如,为了查找包含单  词  Python 的所有日志行.

      你可以这样做:

        lognames = gen_find('access-log*', 'www')
        files = gen_opener(lognames)
        lines = gen_concatenate(files)
        pylines = gen_grep('(?i)python', lines)
        for line in pylines:
            print(line)
        


        如果将来的时候你想扩展管道,你甚至可以在生成器表达式中包装数据。比如,下面这个版本计算出传输的字节数并计算其总和。


        lognames = gen_find('access-log*', 'www')
        files = gen_opener(lognames)
        lines = gen_concatenate(files)
        pylines = gen_grep('(?i)python', lines)
        bytecolumn = (line.rsplit(None,1)[1] for line in pylines)
        bytes = (int(x) for x in bytecolumn if x != '-')
        print('Total', sum(bytes))


        结论


        以管道方式处理数据可以用来解决各类其他问题,包括解析读取实时数据定时轮询等。

        为了理解上述代码,重点是要明白yield 语句作为数据的生产者而 for 循环语句  作为数据的消费者。当这些生成器被连在一起后,每个 yield 会将一个单独的数据元  素传递给迭代处理管道的下一阶段。

        在例子最后部分sum() 函数是最终的程序驱动者,每次从生成器管道中提取出一个元素。这种方式一个非常好的特点是每个生成器函数很小并且都是独立的。这样的话就  很容易编写和维护它们了。

        事实上,由于使用了迭代方式处理,代码运行过程中只需要很小很小  的内存

        在调用 gen_concatenate() 函数的时候你可能会有些不太明白。这个函数的目的  是将输入序列拼接成一个很长的行序列。itertools.chain() 函数同样有类似的功能,  但是它需要将所有可迭代对象最为参数传入。

        在上面这个例子中,你可能会写类似这样  的语句 lines = itertools.chain(*files) ,这将导致 gen_opener() 生成器被提前  全部消费掉。但由于 gen_opener() 生成器每次生成一个打开过的文件,等到下一个迭  代步骤时文件就关闭了,因此 chain() 在这里不能这样使用,,当然上面的方案可以避免这种情况。

        相关文章
        |
        3天前
        |
        前端开发 API
        18_管道——转换
        18_管道——转换
        11 0
        |
        3天前
        |
        Linux Shell
        使用输入输出重定向与管道命令——管道
        使用输入输出重定向与管道命令——管道。
        27 0
        |
        9月前
        |
        Linux
        【进程间通信:管道】(二)
        【进程间通信:管道】(二)
        70 0
        |
        9月前
        |
        消息中间件 Unix Linux
        【进程间通信:管道】(一)
        【进程间通信:管道】(一)
        39 0
        |
        10月前
        |
        消息中间件 Unix Linux
        进程间通信——管道
        管道的基本原理与应用
        |
        11月前
        |
        负载均衡
        进程间通信--管道
        进程间通信--管道
        |
        11月前
        |
        Unix Linux C++
        Linux进程通信——管道(上)
        Linux进程通信——管道(上)
        |
        11月前
        |
        Linux
        Linux进程通信——管道(下)
        Linux进程通信——管道(下)
        线程间的协作(3)——管道输入/输出流
        参考资料《Java并发编程的艺术》《Java编程思想》《Java核心技术卷一》
        1058 0
        |
        分布式数据库 API
        采用管道进行通讯的例子
        作者:肖波    用Remoting做进程间通讯,效率较低,于是做了一个采用管道技术进行进程间通讯的例子,在1.8G 双核计算机上每秒钟可以发送180M数据。下面给出源码       Server端的管道类 using System;using System.
        805 0