Python实现数据库并行读取和写入实例

发布时间 - 2026-01-11 01:48:48    点击率:

这篇主要记录一下如何实现对数据库的并行运算来节省代码运行时间。语言是Python,其他语言思路一样。

前言

一共23w条数据,是之前通过自然语言分析处理过的数据,附一张截图:


要实现对news主体的读取,并且找到其中含有的股票名称,只要发现,就将这支股票和对应的日期、score写入数据库。

显然,几十万条数据要是一条条读写,然后在本机上操作,耗时太久,可行性极低。所以,如何有效并行的读取内容,并且进行操作,最后再写入数据库呢?

并行读取和写入

并行读取:创建N*max_process个进程,对数据库进行读取。读取的时候应该注意:

  1. 每个进程需要分配不同的connection和对应的cursor,否则数据库会报错。
  2. 数据库必须能承受相应的高并发访问(可以手动更改)

实现的时候,如果不在进程里面创建新的connection,就会发生冲突,每个进程拿到权限后,会被下个进程释放,所以汇报出来NoneType Error的错误。

  1. 并行写入:在对数据库进行更改的时候,不可以多进程更改。所以,我们需要根据已有的表,创建max_process-1个同样结构的表用来写入。表的命名规则可以直接在原来基础上加上1,2,3...数字可以通过对max_process取余得到。

此时,对应进程里面先后出现读入的conn(保存消息后关闭)和写入的conn。每个进程对应的表的index就是 主循环中的num对max_process取余(100->4,101->5),这样每个进程只对一个表进行操作了。

部分代码实现

max_process = 16 #最大进程数

def read_SQL_write(r_host,r_port,r_user,r_passwd,r_db,r_charset,w_host,w_port,w_user,w_passwd,w_db,w_charset,cmd,index=None):
  #得到tem字典保存着信息
  try:
    conn = pymysql.Connect(host=r_host, port=r_port, user=r_user, passwd =r_passwd, db =r_db, charset =r_charset)
    cursor = conn.cursor()
    cursor.execute(cmd)
  except Exception as e:
    error = "[-][-]%d fail to connect SQL for reading" % index
    log_error('error.log',error)
    return 
  else:
    tem = cursor.fetchone()
    print('[+][+]%d succeed to connect SQL for reading' % index)
  finally:
    cursor.close()
    conn.close()
  
  try:
    conn = pymysql.Connect(host=w_host, port=w_port, user=w_user, passwd =w_passwd, db =w_db, charset =w_charset)
    cursor = conn.cursor()
    cursor.execute(cmd)
  except Exception as e:
    error = "[-][-]%d fail to connect SQL for writing" % index
    log_error('error.log',error)
    return 
  else:
    print('[+][+]%d succeed to connect SQL for writing' % index)
  
  
  r_dict = dict()
  r_dict['id'] = tem[0]
  r_dict['content_id'] = tem[1]
  r_dict['pub_date'] = tem[2]
  r_dict['title'] = cht_to_chs(tem[3])
  r_dict['title_score'] =tem[4]![](http://images2015.cnblogs.com/blog/1172464/201706/1172464-20170609000900309-1810357590.png)

  r_dict['news_content'] = cht_to_chs(tem[5])
  r_dict['content_score'] = tem[6]
  
  for key in stock_dict.keys():
    #能找到对应的股票
    if stock_dict[key][1] and ( r_dict['title'].find(stock_dict[key][1])!=-1 or r_dict['news_content'].find(stock_dict[key][1])!=-1 ):
      w_dict=dict()
      w_dict['code'] = key
      w_dict['english_name'] = stock_dict[key][0]
      w_dict['cn_name'] = stock_dict[key][1]
      #得到分数
      if r_dict['title_score']:
        w_dict['score']=r_dict['title_score']
      else:
        w_dict['score']=r_dict['content_score']
      
      #开始写入
      try:
        global max_process
        cmd = "INSERT INTO dyx_stock_score%d VALUES ('%s', '%s' , %d , '%s' , '%s' , %.2f );" % \
          (index%max_process ,r_dict['content_id'] ,r_dict['pub_date'] ,w_dict['code'] ,w_dict['english_name'] ,w_dict['cn_name'] ,w_dict['score'])
        cursor.execute(cmd)
        conn.commit()
      except Exception as e:
        error = "  [-]%d fail to write to SQL" % index
        cursor.rollback()
        log_error('error.log',error)
      else:
        print("  [+]%d succeed to write to SQL" % index)

  cursor.close()
  conn.close()
def main():
  num = 238143#数据库查询拿到的总数
  p = None
  for index in range(1,num+1):
    if index%max_process==1:
      if p:
        p.close()
        p.join()
      p = multiprocessing.Pool(max_process)
    r_cmd = ('select id,content_id,pub_date,title,title_score,news_content,content_score from dyx_emotion_analysis where id = %d;' % (index))
    p.apply_async(func = read_SQL_write,args=(r_host,r_port,r_user,r_passwd,r_db,r_charset,w_host,w_port,w_user,w_passwd,w_db,w_charset,r_cmd,index,))

  if p:
    p.close()
    p.join()

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。


# Python数据库并行读取  # Python数据库并行写入  # Python数据库并行  # python实现数据写入excel表格  # Python写入数据到MP3文件中的方法  # Python基于csv模块实现读取与写入csv数据的方法  # python读取excel指定列数据并写入到新的excel方法  # Python3实现将本地JSON大数据文件写入MySQL数据库的方法  # Python实现自定义顺序、排列写入数据到Excel的方法  # Python实现将数据框数据写入mongodb及mysql数据库的方法  # Python实现读写sqlite3数据库并将统计数据写入Excel的方法示例  # 利用python对Excel中的特定数据提取并写入新表的方法  # Python把csv数据写入list和字典类型的变量脚本方法  # Python实现将数据写入netCDF4中的方法示例  # 就会  # 自然语言  # 基础上  # 不可以  # 作了  # 可以通过  # 可以直接  # 在对  # 能找到  # 这篇  # 就将  # 报错  # 这支  # 太久  # 下个  # 只对  # 机上  # 几十万  # 大家多多  # 极低 


相关栏目: 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571


相关推荐: 香港服务器如何优化才能显著提升网站加载速度?  网页设计与网站制作内容,怎样注册网站?  Laravel如何集成Inertia.js与Vue/React?(安装配置)  如何实现建站之星域名转发设置?  javascript日期怎么处理_如何格式化输出  北京网站制作费用多少,建立一个公司网站的费用.有哪些部分,分别要多少钱?  Laravel如何使用Telescope进行调试?(安装和使用教程)  如何在建站之星绑定自定义域名?  手机软键盘弹出时影响布局的解决方法  html5源代码发行怎么设置权限_访问权限控制方法与实践【指南】  BootStrap整体框架之基础布局组件  米侠浏览器网页背景异常怎么办 米侠显示修复  php静态变量怎么调试_php静态变量作用域调试技巧【解答】  中山网站制作网页,中山新生登记系统登记流程?  如何在阿里云通过域名搭建网站?  如何彻底卸载建站之星软件?  微信小程序 scroll-view组件实现列表页实例代码  如何在景安云服务器上绑定域名并配置虚拟主机?  Laravel Telescope怎么调试_使用Laravel Telescope进行应用监控与调试  Laravel如何实现数据库事务?(DB Facade示例)  如何选择PHP开源工具快速搭建网站?  详解Android图表 MPAndroidChart折线图  ,在苏州找工作,上哪个网站比较好?  佛山企业网站制作公司有哪些,沟通100网上服务官网?  C++时间戳转换成日期时间的步骤和示例代码  ,交易猫的商品怎么发布到网站上去?  宙斯浏览器怎么屏蔽图片浏览 节省手机流量使用设置方法  如何用狗爹虚拟主机快速搭建网站?  重庆市网站制作公司,重庆招聘网站哪个好?  Laravel中DTO是什么概念_在Laravel项目中使用数据传输对象(DTO)  Laravel数据库迁移怎么用_Laravel Migration管理数据库结构的正确姿势  Mybatis 中的insertOrUpdate操作  如何自定义建站之星网站的导航菜单样式?  制作电商网页,电商供应链怎么做?  深圳网站制作培训,深圳哪些招聘网站比较好?  Windows10怎样连接蓝牙设备_Windows10蓝牙连接步骤【教程】  齐河建站公司:营销型网站建设与SEO优化双核驱动策略  Android中Textview和图片同行显示(文字超出用省略号,图片自动靠右边)  微信小程序 五星评分(包括半颗星评分)实例代码  Python数据仓库与ETL构建实战_Airflow调度流程详解  如何破解联通资金短缺导致的基站建设难题?  今日头条微视频如何找选题 今日头条微视频找选题技巧【指南】  如何有效防御Web建站篡改攻击?  如何基于云服务器快速搭建网站及云盘系统?  Laravel怎么实现微信登录_Laravel Socialite第三方登录集成  制作无缝贴图网站有哪些,3dmax无缝贴图怎么调?  如何用5美元大硬盘VPS安全高效搭建个人网站?  INTERNET浏览器怎样恢复关闭标签页_INTERNET浏览器标签恢复快捷键与方法【指南】  Laravel怎么处理异常_Laravel自定义异常处理与错误页面教程  Python高阶函数应用_函数作为参数说明【指导】