如何多线程从列表中读取字典并进入数据库

我正在尝试多线程以下代码，但我似乎无法让它工作。

以下代码（我删除了其中的大部分代码只是为了说明目的）目前运行顺利，但速度很慢（3600 条推文列表大约需要 5 分钟）。

import dataset
import datetime
import json
with open("postgresConnecString.txt", 'r') as f:
    DB_CONNECTIONSTRING = f.readline()
DB = dataset.connect(DB_CONNECTIONSTRING)
def load_tweet(tweet, tweets_saved):
    """Takes a tweet (dictionary) and upserts its contents to a PostgreSQL database"""
    try:
        data = {'tweet_id': tweet['tweet_id',
                'tweet_json': json.dumps(tweet)} # Dictionary that contains the data I need from the tweet
        DB['tweets'].upsert(data, ['tweet_id'])
        tweets_saved += 1
        if tweets_saved % 100 == 0:
            print('Saved ' + str(tweets_saved) + ' tweets')
        return tweets_saved
    except KeyError:
        return tweets_saved
if __name__ == "__main__":
    tweets['tweet1', 'tweet2']
    for tweet in tweets:
        tweets_saved = load_tweet(tweet, tweets_saved)

因此，我正在寻找一个选项来执行此多线程。但是，我还没有找到一种方法可以：

多线程提取过程;
、500 或 1000 条推文打印一个计数器;

完成本教程还没有让我理解这样做：每个线程的类的概念，我需要在类中自定义的内容以及目前实现队列对我来说目前有很多需要掌握的;我只是半开始。

有人可以提供有关如何利用多个线程合并上述脚本的提示吗？
我应该使用多少个线程？Python 目前在运行脚本时使用 ~1% 的 CPU 和 ~ 10% 的 RAM（我的系统规格）
我该如何处理递增计数器（使用 Lock（）？），并在达到计数器 % 100 时打印？

编辑：根据要求：以下是分析器结果中的大镜头（带有dataset.upsert）：

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
    5898  245.133    0.042  245.133    0.042 :0(_connect)
    5898   12.137    0.002   12.206    0.002 :0(execute)

这是使用"

dataset.insert"而不是"dataset.upsert"的第二次尝试：

1386332 function calls (1382960 primitive calls) in 137.255 seconds
ncalls  tottime  percall  cumtime  percall filename:lineno(function)
  2955  122.646    0.042  122.646    0.042 :0             (_connect)

最后（绝对并非最不重要的一点），这是运行原始 psycopg2 代码的时间。

63694 function calls (63680 primitive calls) in 2.203 seconds

最后，不要使用数据集来提高性能（尽管编写 psycopg2 代码花了我 10 分钟，>> dataset.upsert 的 10 秒）

现在，至于最初的问题。我是否能够通过多线程进一步减少每个文件的 ~ 2 秒？如何？

完整的代码可以在这里找到

可以改进的几件事：

在单个事务上运行整个批处理。使用事务意味着数据库不需要在每次写入时实际提交（将数据写入磁盘），而是可以在内存上缓冲未提交的数据。这通常会导致更有效的资源使用。

在tweet_id上添加唯一索引。如果没有唯一索引，您可能会强制数据库对每个更新插入执行顺序扫描，这会导致批量更新插入按 O（n**2）缩放。

拆分插入和更新，尽可能使用 .insert_many（）而不是 .upsert（）。在执行批量更新插入之前，请执行预检查询，以查找数据库和推文列表中都存在的tweet_ids列表。使用 .insert_many（）插入数据库中尚不存在的项目，并使用纯 .update（）插入已存在的项目。

我不知道

你是否能够提高性能。但至于我认为你会想要concurrent.futures.Executor.map。ProcessPoolExecutor 而不是 ThreadPoolExecutor 应该是你想要的，尽管我不是专家。

https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor.map

如果要显示进度，请查看同一模块中的concurrent.futures.as_completed。

相关内容

最新更新

热门标签：