如何使用 Flupy 构建数据处理管道

经常使用 Linux 的同学,肯定对|这个符号不陌生,这个符号是 Linux 的管道符号,可以把左边的数据传递给右边。

例如我有一个spider.log文件,我想查看里面包含”ERROR”关键词,同时时间为2019-11-23的数据,那么我可以这样写命令:

1
cat spider.log | grep ERROR | grep "2019-11-23"

但是,如果你想执行更复杂的操作,例如提取关键词fail on: https://xxx.com后面的这个网址,然后对所有获得的网址进行去重,那么虽然 shell 命令也能办到,但写起来却稍显麻烦。

这个时候,你就可以使用 Flupy来实现你的需求。首先我们使用 Python 3.6以上的版本安装Flupy:

1
python3 -m pip install flupy

然后开始写代码,看看这几步操作有多简单:

1
2
3
4
5
6
7
8
9
10
11
12
import re
from flupy import flu

with open('spider.log', encoding='utf-8') as f:
error_url = (flu(f).filter(lambda x: 'ERROR' in x)
.map(lambda x: re.search('fail on: (.*?),', x))
.filter(lambda x: x is not None)
.map(lambda x: x.group(1))
.unique())
for url in error_url:
print(url)

首先flu接收一个可迭代对象,无论是列表还是生成器都可以。然后对里面的每一条数据应用后面的规则。这个过程都是基于生成器实现的,所以不会有内存不足的问题,对于 PB 级别的数据也不在话下。

在上面的例子中,Flupy获取日志文件的每一行内容,首先使用filter进行过滤,只保留包含ERROR字符串的行。然后对这些行通过map方法执行正则表达式,搜索满足fail on: (.*?)\n的内容。由于有些行有,有些行没有,所以这一步返回的数据有些是 None,有些是正则表达式对象,所以进一步再使用filter关键字,把所有返回None的都过滤掉。然后继续使用map关键字,对每一个正则表达式对象获取.group(1)。并把结果输出。

运行效果如下图所示:

实现了数据的提取和去重。并且整个过程通过 Python 实现,代码也比 Shell 简单直观。

由于Flupy可以接收任何可迭代对象,所以传入数据库游标也是没有问题的,例如从 MongoDB 中读取数据并进行处理的一个例子:

1
2
3
4
5
6
7
8
import pymongo
from flupy import flu


handler = pymongo.MongoClient().db.col

cursor = handler.find()
data = flu(cursor).filter(lambda x: x['date'] >= '2019-11-10').map(lambda x: x['text']).take_while(lambda x: 'kingname' in x)

这一段代码的意思是说,从数据库中一行一行检查数据,如果date字段大于2019-11-10就获取text字段的数据,满足一条就获取一条,直到某条数据包含kingname为止。

使用Flupy不仅可以通过写.py文件实现,还可以直接在命令行中执行,例如上面读取spider.log的代码,可以转换为终端命令:

1
flu -f spider.log "_.filter(lambda x: 'ERROR' in x).map(lambda x: re.search('fail on: (.*?),', x)).filter(lambda x: x is not None).map(lambda x: x.group(1)).unique()" -i re

运行效果如下图所示:

通过-i 参数导入不同的库,无论是系统自带的库或者第三方库都可以。

Flupy 的更多使用参数,可以参阅它的官方文档