这样的程序怎么编写？

背景：
有个耗时比较长的程序，每算一条记录，需要几百毫秒。而我要算几百万条。
数据量不大，整个读到内存也就几十兆。
程序计算的结果，要写到数据库。

问题：
由于各种原因，比如网络状况不好，有时候程序会突然崩溃。怎样能做到当程序崩溃再次运行时，能从崩溃的地方开始运行，不用重头开始计算。
现在的程序是每算出一条，就往数据库里写一条数据。（不是我写的）
比如我算到 50 多万条，程序崩溃了，现在的解决办法是删除数据库里面的数据，重新计算，插入。

我的想法是生成一个待计算的任务列表，每计算 1 万条往数据库里写一次，当成功后把这一万条的任务从任务列表里面删除。中间碰到任何情况，把待计算的任务列表写入文件。
下次就可以读取文件中的任务来计算了。

第 1 条附言 · 2016-03-06 18:28:01 +08:00

感谢 @lecher ， @hardware, @ruoyu0088, @firstway, @sololivan, @xujif, @lxy, @winnie2012 ，@lixiaohan, @eliteYang, @tairan2006, @konakona
这里汇报下工作。

现在的处理方式就是用的 try except ，把有错的记录下来，已经在周五解决了。
但是感觉更好的方法是用队列。

其实程序的主体是:

for col1 in df1.columns:
    for col2 in df2.columns:
        ols_result = pandas.ols(df1[col1], df[col2])
        result_needed = get_what_we_need(ols_result)
        write_to_database(result_needed)

ols 不仅会计算很多东西，而且，每次都会对数据进行清理，比如对齐索引，去掉 NA 值。

我现在的做法，
一是自己写 OLS ，只计算我需要的结果。

第二就是清理数据。
由于我是在前人的结构上改的，没有大动代码的结构。
考虑到 df1 里面的 col1 会和 df2 里面的每一列都运算，而且经过 line_profiler 测试，程序 dropna()是很耗时的，所以写了一个 cache 装饰器，用来保存 dropna()后的列。

第三把程序改成并行的。
由于 df1 有 2000+列， df2 也一样，我是先做出一个列表 job_lst = itertools.product(df1.columns, df2.columns)
然后对 job_lst 分批次进行并行计算。

这样改下来后，程序从原来的需要几十个小时到只需要不到四个小时。

再次感谢大家，这里就当做自己做个记录，也给后来人提供个经验 /教训吧。

计算

程序

数据库

崩溃

23 条回复 • 2016-03-04 16:15:33 +08:00

jugelizi

2016-03-03 23:07:37 +08:00

爬虫么

yixiang

2016-03-03 23:16:31 +08:00

每次开始时根据数据库里的数据，判断上次进行到了哪里，从那里开始。

crystom

2016-03-03 23:20:26 +08:00

说下我最简单的想法数据数据库都分成几部分每次执行一部分的计算

zhjits

2016-03-03 23:21:16 +08:00

参考文件系统日志的实现

cgcs

2016-03-03 23:21:43 +08:00

网络不好->程序奔溃~~~什么鬼

爬虫的话，不大可能几百毫秒完成一个吧

这种程序，丢到后台去计算就是了~~~

billgreen1

2016-03-03 23:25:38 +08:00

是数据太脏了，考虑了很多情况，但是有时后还是会出现问题。

billgreen1

2016-03-03 23:30:44 +08:00

我就是对两列数据进行线性回归，使用的是 pandas 的 Series, 经常出现序列 a 和序列 b 不一样长，我每次要根据 a 和 b 的 index 合并数据，并且去掉 NA ，然后进行计算，有时候整个 a 或 b 都是 na ，更扯的是，有时候根本取不到 a 或 b ， a / b 是从一个 dataframe 里面取的. 我期望有个列名比如叫 colA, colB, 有时候没有，这时候会报 IndexError.

lecher

2016-03-03 23:35:04 +08:00

任务没有切分好职责。步骤也没有拆分好。
1.假设计算完写入一条数据，那么这条数据的索引是什么？下一次要更新这条数据需要拿到那些信息才能查询到？理清楚这个对任务的管理非常重要，只要把这个数据的索引确定了，至少不用删除重新计算。
2.存入内存的基础数据，进行计算时会产生哪些数据结构和数据？如果没有使用内存缓存 redis 、 memcache 之类的，可以使用写入文件的方式将数据结构和数据序列化到文件中，这样下一次读取数据不需要重新建模，可以直接计算。
3.计算的步骤能不能切分成几个无上下文状态的任务，将计算步骤拆开，每一个步骤只负责读取数据进行计算，计算之后将结果写入 Queue 或者文件，提供给另外一个步骤的读取处理，这样可以更方便存储不同步骤产生的缓存数据，方便出现异常时进行恢复。

lecher

2016-03-03 23:44:25 +08:00

看起来写好一个洗数据的任务，专门将异常数据挑出来，先将洗过的基础数据序列化存到数据库或者文件，预留一个处理标识。
现有的计算业务读取洗过的数据，先检测处理标识是否处理过，没有处理过则交给计算业务处理，处理过则跳过。
这样不需要对现有代码进行大的修改，只要把洗数据的异常处理完善，处理的状态也不会因为程序崩溃而丢失太多。

hardware

2016-03-04 02:05:07 +08:00

用 Python 的话定时用 pickle 保存一下参数好了很简单
我们的运算量比你这个大成千上万倍…

ruoyu0088

2016-03-04 06:41:03 +08:00

要是我做的话，我就来一个大的 try except ，保证程序不崩溃。遇到不能处理的数据就单独输出到文件中，等能处理的数据都处理完毕之后。再查看文件研究数据中有何不能处理的问题。改进程序，如此重复。直到所有的数据处理完毕。

另外， Pandas 如果使用不得当会很慢的，你确定的数据处理程序已经优化好了么。

firstway

2016-03-04 07:45:38 +08:00 via Android

如果我来做，我会写 3 个程序。
一个 dump ，从数据库读出来，写到本地文件，比如最简单的文本。一行一个记录或类似。
第二个处理程序，处理同时记录处理那些到临时文件，相当于前面人提的日志。崩溃后自动读取日志，从上次中断地方重新开始。
第 3 个程序就把结果写回数据库。

看似 3 个程序，很麻烦的样子，其实反而不复杂。
而且第二程序可以以后用在其他地方，因为它的输入输出是本地文件，跟数据没关系。
而且第一第三可能用 SQL 就可以搞定。所以 3 个东西可以几个人同时做。

而且如果想做，可以 3 个程序 pipeline 起来。参见 Unix 管道。
anyway ，都是 KlSS 哲学的一些应用。

firstway

2016-03-04 07:46:58 +08:00 via Android

“跟数据没关系”，应该是“跟数据库没关系”

firstway

2016-03-04 07:50:13 +08:00 via Android

而且因为有本地输入输出文件，借助 vi ， diff ， grep ， awk 来 debug ，异常方便的直观。

sololivan

2016-03-04 09:19:59 +08:00

消息队列

lxy

2016-03-04 09:30:04 +08:00

计算密集直接上 C ，动态调用

xujif

2016-03-04 12:00:11 +08:00

消息队列+1 没有更合适的了

winnie2012

2016-03-04 12:07:36 +08:00

分拆任务和最小化接口操作，同意 @firstway 的方案，请楼主执行。

wodemyworld

2016-03-04 12:14:41 +08:00

数据库服务器上写个信息接收程序，校验 md5 、 sha1 码，一致的话就这个本地程序写入数据库，网络不好还不解决网络问题啊、、、、、、

lixiaohan

2016-03-04 15:42:10 +08:00

@sololivan 消息队列挺好的比如 rabbitmq 你把所有要处理的东西都放到 rabbitmq 中，用 gevent 并行去处理就好了

eliteYang

2016-03-04 15:49:56 +08:00

先把程序会崩溃解决了，然后再考虑其他方案，可以放进消息队列里，或者处理过的放入数据库里。
最重要的是把任务分拆成最小的，然后再处理，就算崩了也知道处理到了哪里

tairan2006

2016-03-04 16:07:26 +08:00

把偏移量记下来不就完了。。

当然用消息队列更好。

konakona

2016-03-04 16:15:33 +08:00

必須聯網運行嗎？不過不管聯網與否，數據庫的依賴是必不可少的。

可以配合 Redis （或類似工具）來做入庫的中間件工作，這樣做的好處是減輕計算程序的依賴性。
程序首次執行，把 100 個（這個量你來定）放在剛好能容納 100 個空間的 Redis ，每處理完一個再插一個。

負責計算的程序可以多線程、多進程嗎？可以準備守護進程一直檢查這個 name 的進程是否還存活，如果掛了可以再開啟。

這樣做的好處是分工明確減輕計算程序的壓力和複雜度。