您好, 欢迎来到 !    登录 | 注册 | | 设为首页 | 收藏本站

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

5b51 2022/1/14 8:16:52 python 字数 7208 阅读 331 来源 www.jb51.cc/python

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

概述

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

前言

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

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

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

并行读取和写入

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

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

此时,对应进程里面先后出现读入的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,%.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_cmd,index,))

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

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

总结

以上是编程之家为你收集整理的Python实现数据库并行读取和写入实例全部内容,希望文章能够帮你解决Python实现数据库并行读取和写入实例所遇到的程序开发问题。


如果您也喜欢它,动动您的小指点个赞吧

除非注明,文章均由 laddyq.com 整理发布,欢迎转载。

转载请注明:
链接:http://laddyq.com
来源:laddyq.com
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。


联系我
置顶