- 加速处理大数据的思路
- 动机
- 最开始的方法
- 1. 概述
- 2. 遇到的问题
- 3. 速度慢的根本原因
- 优化后的方法
- 1. 概述
- 2. 具体方法（具体代码看下一章）
- 方法一：批量查询数据，减少调用数据库的次数
- 方法二：建立数据库索引并定时重建索引
- 方法三：查询数据时指定列，不要全部查询所有列
- 方法四：多进程运行python程序
- 方法五：用`DataX`工具 将结果存入数据库
- 推荐方法/工具
- multiprocessing：多进程运行python程序
- 批量查询、批量更新：减少调用数据库的次数
- DataX：传输大量数据
- 个人总结
频繁调用数据库会耗费大量时间，所以需要减少调用数据库次数。
此外，可以借助DataX工具 将数据存入数据库。
动机我要用python标识几千万条存放在sql server数据库中的数据，并将结果作为新的列插入到原表中。工作可以分为三个部分：
① 从数据库中取数据
② 用python给数据打标签
③ 将标签更新到数据库
如图所示：
最开始我一条一条地处理数据，并且处理完就马上存入数据库。但是这样不仅程序运行特别慢，还经常因为频繁调用数据库导致报错。
所以我就一直摸索加速程序运行的方法，最终将程序运行速度提升了100万倍！
-
之前的程序：一天最多能处理80000多条数据。
-
现在的程序：一天能处理9000万条数据。
从数据库中一条一条取数据（①），取一条数据就处理一条（②），处理完马上将结果存入数据库（③）。如下图：
2. 遇到的问题-
处理数据速度特别慢，平均一秒只能处理1~2条数据。一天大概只能处理8万条数据。
-
对数据库进行查询操作或更新操作时经常会报错：
pymssql.OperationalError: (20009, b'DB-Lib error message 20009, severity 9:nUnable to connect: Adaptive Server is unavailable or does not exist ...
我几乎试了全网所有的解决方法都没办法解决这个问题，后来问了公司的前辈，前辈说可能是因为太过频繁调用数据库，占用太多资源。然后他让我用union语句拼接查询结果，这样就能做到批量查询（下章会详细介绍）。
3. 速度慢的根本原因经过分块测试后，我发现程序运行速度慢的根本原因是频繁调用数据库，并且更新数据库比查询更耗时。
基于此，我下面采取了几个手段来加速程序运行。
优化后的方法 1. 概述首先我一次性将所有的数据都从数据库中取出来（①），然后多进程运行python程序处理数据（②），再将所有的结果存入csv文件中（③），然后用DataX工具 将csv导入数据库（④），最后合并原数据Data Table和标签数据Tag Table（⑤），如图：
相较于之前的方法，现在的方法优化了以下几点：
- 优化查询和更新数据方法；
- 多进程运行python程序。
- 按批次拼接查询语句，并进行查询；
- 将每次的查询结果都存在列表里。
- 将Data Table的id设置为主键（注意要先去重，因为主键是唯一的）
alter table test_table add primary key(id)
-
在Microsoft Sql Server Management Studio中设置定时作业，定时重建索引
不定时重建索引会越跑越慢，具体什么原因我不清楚，如果想知道你们自己去查。
重建索引代码：
alter index pk_asdguy1g12uy3g123 on test_table rebuild -- pk_asdguy1g12uy3g123是test_table的主键
设置定时作业的步骤：
-
第一步：
-
第二步：
-
第三步：
-
第四步：
-
第五步：
-
第六步：
-
第七步：
-
第八步：重要！！！
-
第九步：
-
比如我需要每个数据的id，我只需要：
select id from test_table
而不要：
select * from test_table方法四：多进程运行python程序
利用python的multiprocessing模块，设置4个进程并行运行程序。
方法五：用DataX工具 将结果存入数据库-
将生成的tag存入csv文件
-
然后用DataX工具将csv文件存入数据库
-
最后再合并原表和tag表
-
python代码：
from multiprocessing import Pool def fun1(n): for i in range(10): print(i+n) if __name__ == '__main__': process_number = 4 # 设置进程数 p = Pool(process_number) # 下面分了process_number个进程 p.apply_async(fun1(1)) # 将进程加入进程池 p.apply_async(fun1(10)) p.apply_async(fun1(100)) p.apply_async(fun1(1000)) p.close() p.join()注意：
- 要将多进程代码写在if __name__ == '__main__':函数里；
- 进程数可以超过cpu核数，但最好将进程数设置小于或等于cpu核数，不然电脑会很卡。
-
推荐教程：Python开启多个进程
-
批量查询
-
假设test_table内容为：
id data 1111 data1 2222 data2 3333 data3 4444 data4 -
sql代码：
select * from test_table where id in (1111, 2222, 3333, 4444) -- 查询id为1111, 2222, 3333, 4444的所有数据 order by CHARINDEX(ltrim(id), '1111, 2222, 3333, 4444') -- 并将结果按照1111, 2222, 3333, 4444进行排序
-
python代码：最大查询数量不要超过20000，因为超过20000，会因为sql语句太长而报错
import math import tqdm from utils.util_function import execute_query # 执行查询语句 def get_all_datas(process_name, DBname, data_ids): """ :param process_name:进程数 :param DBname: 数据库名字 :param data_ids: 数据id：[[data_id, index], ...] :return: 以字典形式返回 """ data_length = len(data_ids) max_length = 20000 result_dict = {} m = math.ceil(data_length / max_length) # int()是向上取整 pre_index = next_index = 0 for i in range(m): """ 拼接查询语句 """ sql_string = "select * from {} with(nolock) where id in (".format(DBname) pre_index = next_index # 更新index if i < m-1: next_index = pre_index + max_length else: next_index = data_length # 获取这一批的数据 smaller_list = data_ids[pre_index:next_index] length = next_index - pre_index for j in tqdm(range(length), desc=process_name + ': Splicing ' + str(i+1) + 'st ' + 'string for datas'): if j != 0: sql_string += ", " data_id = smaller_list[j][0] sql_string += str(data_id) sql_string += ") order by CHARINDEX(ltrim(id), '" for j in tqdm(range(length), desc=process_name + ': Splicing ' + str(i+1) + 'st ' + 'string for datas'): if j != 0: sql_string += ", " data_id = smaller_list[j][0] sql_string += str(data_id) sql_string += "')" """ 从数据库读取数据 """ datas = execute_query(sql_string) # 执行查询语句 """ 将数据按{data_id : data}分组存入字典 """ for j in tqdm(range(length), desc=process_name + ': Storing in list to ' + str(i+1) + 'st list for datas.'): data_id = smaller_list[j][0] value = datas[j] result_dict[data_id] = value return result_dict
-
-
批量更新
如果数据量较少可以用批量更新的方法，否则就用上述方法四，不然还是会报错
-
假设更新前的test_table内容为：
id data tag 1111 data1 0 2222 data2 0 3333 data3 0 4444 data4 0 -
sql代码：
UPDATe test_table SET tag = CASE id when 1111 then -1, when 2222 then 1, when 3333 then -1, when 4444 then 1 END WHERe id in (1111, 2222, 3333, 4444)
-
python代码：最大查询数量不要超过500，因为超过500，会因为sql语句太长而报错
import math import tqdm from utils.util_function import execute_non_query # 执行非查询语句 def batch_tag_datas(tag_list, DBname, column_name): """ 批量更新tag列 :param tag_list: 标签列表：[[data_id, tag], ...] :param DBname: 数据库名称 :param column_name: 列名 :return: """ max_length = 500 data_num = len(tag_list) pre_index = next_index = 0 m = math.ceil(data_num / max_length) # int()是向上取整 print("--------- Start update tag ---------") for i in tqdm(range(m), desc='Update tag in batch: '): # 分批处理 """ 拼接更新语句 """ pre_index = next_index if i < m - 1: next_index = pre_index + max_length else: next_index = data_num small_list = tag_list[pre_index:next_index] length = next_index - pre_index sql_string = "UPDATE {} SET {} = CASE id ".format(DBname, column_name) updated_ids = "(" for j in range(length): data_id = small_list[j][0] tag = small_list[j][1] temp_string = "when {} THEN {} ".format(data_id, tag) if j < length - 1: sql_string += temp_string updated_ids += (str(data_id) + ", ") else: temp_string += "END WHERe id in " sql_string += temp_string updated_ids += (str(data_id) + ")") sql_string += updated_ids """ 更新数据 """ execute_non_query(sql_string) -
更新后结果：
id data tag 1111 data1 -1 2222 data2 1 3333 data3 -1 4444 data4 1
-
- Github：
- DataX工具
- DataX SqlServerWriter 介绍
- 推荐教程：
- datax安装需要的环境
- 学习 Python3 使用 DataX 读取 csv 文件，写入另一个 csv 文件
- datax介绍及生产脚本配置
之前我嫌麻烦，想只用python就解决所有问题，并且很抵触去改代码，导致项目进度停滞很久。但其实很多东西用sql或其他工具更快。
后来我沉下心大改了代码，并且放弃用python处理一些东西，而是去学了别的工具怎么用，发现用别的工具比用python快多了。
还有，沟通真的很重要。之前疫情被封在家里3个月，我几乎没怎么和别人交流，就只是自己写代码，导致进度很慢。
后来解封后，我积极地向上司和技术部的同事寻求帮助，然后我的项目进展突飞猛进，并且我也学到很多新东西。
比如说之前我想将有1000万行数据的csv文件导入数据库，我之前使用python一条一条写入数据库，大概要花半年才能写完，但是技术部的老师10分钟就帮我导入了。
暑期编程PK赛 得CSDN机械键盘等精美礼品！
