记一次小机器的Python大数据分析

0x00 前言

机缘巧合，公司突然要搞一波大量数据的分析。属于客流类的分析。

数据量级也还算不错，经过 gzip 压缩，接近 400 个 点位的 SQL 文件 (MySQL innoDB)，大小接近 100GB 左右，原始记录数据估测在 180 亿左右。

解压后...... 差不多一个 T 吧。

如果是人民币玩家，自然是直接购置几十台高配置机器，做个 mysql shard 或者直接上大数据全家桶比如 hadoop 和 hive 之类，让程序员去往死里折腾吧。

嗯，然而对于我这种非人民币玩家，就要用单机硬扛。

那就硬扛呗。

我手上的机器配置如下：

• 局域网服务器 （ Ubuntu 16.04 LTS ）

  • Xeon(R) CPU E3-1225 v5 @ 3.30GHz
  • 16G 内存
  • 1T 硬盘

• 苹果电脑 2016 年 15 寸 最高配

  • 1T 硬盘
  • i7 四核

0x01 准备数据阶段

用低配机器分析大数据的首要原则，就是不要分析大数据。

何也？

就是尽可能的抽取所得结论所需分析数据的最小超集

小机器是无法完成海量计算的，但通过一定的过滤和筛选可以将数据筛选出到一台机器能扛得住的计算量。从而达到可以可以分析海量数据的目的。

1.1 将数据导入 MySQL 中

我们先不管三七二十一，既然给了 SQL 文件，肯定要入库的，那么问题来了：

将大象关进冰箱要几个步骤

将数据导入数据库中需要几个步骤

或者说，如何更快的导入 400 张不同表的数据。

大致步骤如下：

• 新增硬盘，并初始化
• 配置 MySQL 的 datadir 到新增硬盘上
• 导入数据 (PV & MySQL)

新增硬盘，并初始化

首先，购买并插入硬盘

使用 lshw 查看硬盘信息

root@ubuntu:~# lshw -C disk
  *-disk
       description: SCSI Disk
       product: My Passport 25E2
       vendor: WD
       physical id: 0.0.0
       bus info: scsi@7:0.0.0
       logical name: /dev/sdb
       version: 4004
       serial: WX888888HALK
       size: 3725GiB (4TB)
       capabilities: gpt-1.00 partitioned partitioned:gpt
       configuration: ansiversion=6 guid=88e88888-422d-49f0-9ba9-221db75fe4b4 logicalsectorsize=512 sectorsize=4096
  *-disk
       description: ATA Disk
       product: WDC WD10EZEX-08W
       vendor: Western Digital
       physical id: 0.0.0
       bus info: scsi@0:0.0.0
       logical name: /dev/sda
       version: 1A01
       serial: WD-WC888888888U
       size: 931GiB (1TB)
       capabilities: partitioned partitioned:dos
       configuration: ansiversion=5 logicalsectorsize=512 sectorsize=4096 signature=f1b42036
  *-cdrom
       description: DVD reader
       product: DVDROM DH1XXX8SH
       vendor: PLDS
       physical id: 0.0.0
       bus info: scsi@5:0.0.0
       logical name: /dev/cdrom
       logical name: /dev/dvd
       logical name: /dev/sr0
       version: ML31
       capabilities: removable audio dvd
       configuration: ansiversion=5 status=nodisc
复制代码

使用 fdisk 格式化硬盘，并且分区

fdisk /dev/sdb
#输入 n
#输入 p
#输入 1
#输入 w
sudo mkfs -t ext4 /dev/sdb1
mkdir -p /media/mynewdrive
vim /etc/fstab
# /dev/sdb1    /media/mynewdrive   ext4    defaults     0        2
# 直接挂载所有，或者 reboot
mount -a
复制代码

至此为止，硬盘就格式化完成了。

关于安装硬盘，可以参考 https://help.ubuntu.com/community/InstallingANewHardDrive

配置 MySQL

篇幅有限，只简介具体在 Ubuntu 16.04 上面 配置 MySQL 的 DataDIR ，省去安装和基本登录认证的配置。

mysql 在 ubuntu 下面默认的路径如下：

/var/lib/mysql/
复制代码

我们开始配置 DataDIR

systemctl stop mysql
rsync -av /var/lib/mysql /mnt/volume-nyc1-01
mv /var/lib/mysql /var/lib/mysql.bak
vim /etc/mysql/mysql.conf.d/mysqld.cnf
# 修改至 datadir=/mnt/volume-nyc1-01/mysql
vim /etc/apparmor.d/tunables/alias
# alias /var/lib/mysql/ -> /mnt/volume-nyc1-01/mysql/
sudo systemctl restart apparmor
vim /usr/share/mysql/mysql-systemd-start
# 修改成
if [ ! -d /var/lib/mysql ] && [ ! -L /var/lib/mysql ]; then
 echo "MySQL data dir not found at /var/lib/mysql. Please create one."
 exit 1
fi

if [ ! -d /var/lib/mysql/mysql ] && [ ! -L /var/lib/mysql/mysql ]; then
 echo "MySQL system database not found. Please run mysql_install_db tool."
 exit 1
fi

# 接下来
sudo mkdir /var/lib/mysql/mysql -p
sudo systemctl restart mysql

# 最后 my.conf 修改相关文件路径
复制代码

详细请参考这篇文章 https://www.digitalocean.com/community/tutorials/how-to-move-a-mysql-data-directory-to-a-new-location-on-ubuntu-16-04

将 DataDIR 配置完成之后，就可以导入数据了。嗯，经过这么麻烦的事情之后，我决定下次遇到这种情况首选 Docker 而不是在 Ubuntu Server 上面搞这个。

站在现在看，如果重来的话，我肯定会用 Docker 然后把数据盘挂载到新硬盘到。

比如直接 Docker 命令执行

docker run --name some-mysql -v /my/own/datadir:/var/lib/mysql -e MYSQL_ROOT_PASSWORD=my-secret-pw -d mysql:tag
复制代码

导入数据 之 MySQL + PV

我们使用 mysql 导入脚本的时候，有几种导入方式

• source 命令，然而这个命令容易在数据量很大的时候直接卡掉。（印象中是直接把 sql 文件加载到内存中，然后执行，然而，只要涉及到大量文本打印出来并且执行，速度一定会变慢很多）
• mysql 命令

# mysql 命令的典型导入场景就是这样
mysql -uadmin -p123456 some_db < tb.sql
复制代码

加上 PV 命令的话，比较神奇了。有进度条了！!

# 附加进度条的导入场景
pv -i 1 -p -t -e ./xxxx_probe.sql | mysql -uadmin -p123456 some_db
复制代码

然后，可以查看一下磁盘 CPU 内存的占用情况。如果负载（着重注意 IO，内存）还不够满，使用 tmux 多开几个进程导入数据。

因为每个 SQL 文件对应的表不一样，所以多开几个进程批量 insert 的话并不会锁表，这样可以显著提升导入速度。

1.2 导出数据

既然已经导入了数据，为什么需要导出数据呢？

因为数据量比较大，需要进行初步清洗。而我们最后肯定使用 Pandas 进行分析，从局域网数据库中读取大量的数据的时候，pandas 速度会非常的慢（具体是因为网络传输速度？)。所以，为了后面分析省事，我批量导出了数据，然后按照我的习惯进行了归类。

在这个过程中，我还进行了一小部分的数据过滤，比如：

• 只选取对自己有用的行与列。
• 化整为零，拆分数据为最小单元的 CSV 文件

只选取对自己有用的行与列

select col_a , col_b from some_table where Acondition and bcondition and col_c in ('xx','yy','zz');
复制代码

这里面有一些值得注意的地方

• 尽量把简单的判断写在左边。
• 如果不是反复查询，则没有必要建立索引。直接走全表，筛选出必要的数据存 CSV 即可。

尽量拆分数据为最小单元的 CSV 文件

如果按照某类，某段时间进行拆分可以在分析的时候随时取随时分析那就进行拆分。

比如，某个大的 CSV 包含琼瑶里面各种人物情节地点的位置就可以拆分为：

201712_大明湖畔_夏雨荷_还珠格格_你还记得吗.csv
201711_老街_可云_情深深雨蒙蒙_谁来救我.csv
201710_屋子里_云帆_又见一帘幽梦_你的腿不及紫菱的爱情.csv
复制代码

当我们需要取这坨数据的时候，可以直接 glob 一下，然后 sort, 接着二分查找。就可以快速读取这块数据了。

1.3 校验数据完备性

第三方给的数据多多少少会有这些或者那些的问题，一般情况下，可以通过检查数据完备性来尽可能的减少数据的不靠谱性。

我习惯性在这样的表里面详细记录数据完备性的各种参数与进度。

比如：

• 数据的提供情况和实际情况
• 阶段性的记录条数和点位的统计值
• max，min，mean，median 用来避免异常值
• 如果是分年份，则必须要统计每一天的情况，否则也不知道数据的缺失程度。

0x02 分析阶段

经过上一步处理，数据的文件总大小大约从 1000GB (uncompressed) -> 30GB 左右 （拆分成若干个文件 compressed) 。每个文件大约是几百兆。

2.1 性能要点 1：文件系统

如果统计逻辑很简单，但是数量多，首选使用读取文件。读取文件进行统计速度是非常快的。（人民币玩家走开）

像 linux 里面的 wc,grep,sort,uniq 在这种场景有时候也能用到。

但，注意，如果文件特别大，一定要迭代器一个一个读取。

2.2 性能要点 2：化整为零，map reduce filter

化整为零这个已经在上面的 1.2 节讲过了。

map/reduce/filter 可以极大的减少代码。

collection 中有个 Counter , 在进行简单代码统计的时候用起来可以极大的减少代码。

2.3 性能要点 3：进程池的两种作用

我们都知道，当 用 Python 执行计算密集的任务时，可以考虑使用多进程来加速：

即为了加速计算，此为作用一。如下：

def per_item_calc(item):
    df = pd.read.....
    # complex calc
    return result

with ProcessPoolExecutor(3) as pool:
    result_items = pool.map(per_item_calc,all_tobe_calc_items)

reduce_results = ....
复制代码

其实进程的销毁本身就可以给我带来第二个作用管理内存。

具体会在 2.6 中的 DataFrame 里面解释。

2.4 性能要点 4：List 和 Set , itertools

有 400 组 UUID 集合，每个列表数量在 1000000 左右，列表和列表之间重复部分并不是很大。我想拿到去重之后的所有 UUID，应该怎么处理

在去重的时候，自然而然想到了使用集合来处理。

最初的做法是

list_of_uuid_set = [ set1 , set2 ... set400 ]
all_uuid_set = reduce(lambda x: x | y, list_of_uuid_set)
复制代码

1 小时过去了。 突然之间，四下里万籁无声。公司内外聚集数百之众，竟不约而同的谁都没有出声，便有人想说话的，也为这寂静的气氛所慑，话到嘴边都缩了回去。似乎硬盘的指示灯也熄灭了，发出轻柔异常的声音。我心中忽想：

小师妹这时候不知在干甚么？ 卧槽，程序是不是又卡死了？

SSH 上去 htop 一下机器。发现实存和内存都满了。直觉告诉我，CPython 的集合运算应该是挺耗内存的。

嗯，这怎么行，试试用列表吧。列表占用内存应该是比较小的。

def merge(list1,list2):
    list1.append(list2)
    return list1

list_of_uuid_list = [ list1 , list2 ... list400 ]
all_uuid_set = set(reduce(merge, list_of_uuid_list))
复制代码

1 小时过去了。 我一拍大腿，道：

小师妹这时候不知在干甚么？ 卧槽，程序是不是又卡死了？

最后在 StackOverFlow 上找到了更好的解决方案。

list_of_uuid_list = [ list1 , list2 ... list400 ]
all_uuid_set = set(list(itertools.chain(*list_of_uuid_list)))
复制代码

运行一下，5s 不到出了结果（注意，包含了 Set 去重）。

itertools 里还有很多有趣的函数可以使用。

https://docs.python.org/3/library/itertools.html

2.5 性能要点 5：IPython 给性能带来的影响

当我们在分析数据的时候，往往使用的是 IPython, 或者 Jupyter Notebook

但是，方便的同时，如果不加以注意的话，就会带来一点点小问题。

比如下划线和双下划线分别存储上一个 CELL 的返回值，和上上个 CELL 的返回值。

2.6 性能要点 6：DataFrame 带来的 GC 问题

DataFrame 是我用 Pandas 的原因，在这次使用 DataFrame 的过程中，还是出现一些头疼的问题的。比如莫名的内存泄露。

def per_item_calc(item):
    df = pd.read.....
    # complex calc
    return result

result_items = []
for item in all_tobe_calc_items:
    result_items.append(per_item_calc(item))

reduce_results = ....
复制代码

我在 For 循环中读取 DataFrame 赋值给 df, 然后统计出一个结果。按理来说，每次只要一个简单的 result, 每次读取的文件大小一致，同样的会占用接近 2G 内存，而，当我赋值 df 的时候，按理来说，应该是把原先 df 的引用数应该为 0, 会被 gc 掉，又释放了 2G 内存，所以，是不太可能出现内存不够用的。

运行程序，内存 biubiubiubiu 的增长，当进行到约第 1000 次的循坏的时候，直到 16G 内存占满。

那么显式的 del 一下会不会好一点呢？代码如下：

def per_item_calc(item):
    df = pd.read.....
    # complex calc
    del df
    return result
复制代码

似乎好了一点点，但是其实并没有好到哪里去。

然而，和前一次一样，内存 biubiubiubiu 的增长，当进行到约第 1000 次的循坏的时候，直到 16G 内存占满。

只是在读取文件的时候，预先减少了上次循环没有 del 掉的 df. 和上一个想法没有太大区别。除了比上一个方法每次读取文件的提前减少了一个多 G 的内存。

查找相关资料，涉及到 Python 里面的 Pandas GC 的资料并不多，稍微整理一下，如下：

Python 程序 在 Linux 或者 Mac 中，哪怕是 del 这个对象，Python 依旧 站着茅坑不拉屎 就是不把内存还给系统，自己先占着，有本事你打死我啊 直到进程销毁。

嗯？这个和我要的东西不一样嘛？具体怎么管理 pandas 里面的 object 的，到底是哪里 GC 不到位呢？还是没有说呀。

参考：

• https://stackoverflow.com/questions/23183958/python-memory-management-dictionary
• http://effbot.org/pyfaq/why-doesnt-python-release-the-memory-when-i-delete-a-large-object.htm

不过有一点启示了我。

直到进程销毁。

Python 里面不是有个 ProcessPoolExecutor 模块么。

那么问题来了，ProcessPoolExecutor 是动态创建进程并且分配任务的呢，为每一个 item 分配一个进程来运算？还是创建完三个进程之后把 item 分配给空闲进程的进行运算呢？

• 如果是后者，则是正经的进程池。似乎 map 过去，除非任务执行完毕或者异常退出，否则进程不销毁。并不能给我们解决 内存泄露 的问题。
• 如果是前者，则是并不是线程池，但是可以帮我解决内存泄露的问题。

你说，进程池肯定是前者咯。可是你在验证之前，这是进程池只是你的从其他语言带来的想法，这是不是一个线程池，是一个什么样子的进程池，如果进程执行过程中挂掉了，这个时候就少了一个线程，会不会再补充一个进程呢？？

怎么看验证呢？

1. 运行程序，进入 Htop 看进程 PID
2. 看源码

# https://github.com/python/cpython/blob/3.6/Lib/concurrent/futures/process.py#L440
def _adjust_process_count(self):
    for _ in range(len(self._processes), self._max_workers):
        p = multiprocessing.Process(
                target=_process_worker,
                args=(self._call_queue,
                        self._result_queue))
        p.start()
        self._processes[p.pid] = p
复制代码

从源码得出在主线程创建了管理进程的线程，管理进程的线程创建了 max_workers 个进程（在我的例子里面就只有 3 个 worker).

是个进程池。

好，如果是进程池，似乎 map 过去，除非任务执行完毕或者异常退出，否则进程不销毁。并不能给我们解决 内存泄露 的问题。

等等，如果用多进程池不就好咯？

def per_item_calc(item):
    df = pd.read.....
    # complex calc
    return result

result_items = []
step = 300
for idx in range(0,len(all_tobe_calc_items),step):
    pieces_tobe_calc_items = all_tobe_calc_items[idx:idx+step]
    with ProcessPoolExecutor(3) as pool:
        pieces_result_items = pool.map(per_item_calc,pieces_tobe_calc_items)
        result_items.append(pieces_result_items)

reduce_results = list(itertools.chain(*result_items))
复制代码

当然，这是一种让操作系统帮我 GC 的方法。即 Python 不能帮我 GC 的，操作系统帮我 GC

PS: 其实用 multiprocessing 模块也行，只是线程池可以稍微控制一下进程创建的数量。

总结一下，对于大量的 DataFrame 处理：

1. 多个进程池是一种处理的方式。
2. 尽量减少 DataFrame 的数量
3. 尽量减少赋值导致的 COPY, 修改时带上 inplace=True
4. 读取 CSV 的时候指定相关列的类型 {‘col_a’: np.float64, ‘col_b’: np.int32}，否则 pandas 会产生大量的 object

0xDD 番外篇

在分析这次的数据过程中，自己的 Mac 主板也坏掉了，幸好还在保修期，送到苹果店维修了一下。给苹果的售后点个赞。

0xEE 更新

• 2017-12-07 初始化本文
• 2017-12-16 增加分析阶段的文字
• 2017-12-26 去掉一些 TODO, 发布到我的小站
• 2017-12-31 正式发布