我想基于a中的值创建一个类的实例pandas.DataFrame.我已经失败了.

import itertools
import multiprocessing as mp
import pandas as pd

class Toy:
    id_iter = itertools.count(1)

    def __init__(self, row):
        self.id = self.id_iter.next()
        self.type = row['type']

if __name__ == "__main__":

    table = pd.DataFrame({
        'type': ['a', 'b', 'c'],
        'number': [5000, 4000, 30000]
        })

    for index, row in table.iterrows():
        [Toy(row) for _ in range(row['number'])]

多处理尝试

我已经能够通过添加以下内容来并行化这种(某种程度):

pool = mp.Pool(processes=mp.cpu_count())
m = mp.Manager()
q = m.Queue()

for index, row in table.iterrows():
    pool.apply_async([Toy(row) for _ in range(row['number'])])

如果数字row['number']显着长于长度,这似乎会更快table.但在我的实际情况中,table …

我有一个存储对象的字典:

jobs = {}
job = Job()
jobs[job.name] = job

现在我想将它转换为使用管理器dict,因为我想使用多处理并需要共享这个dict amonst进程

mgr = multiprocessing.Manager()
jobs = mgr.dict()
job = Job()
jobs[job.name] = job

只是通过转换为使用manager.dict(),事情变得非常缓慢.

例如,如果使用原生字典,则只需0.65秒即可创建625个对象并将其存储到dict中.

同样的任务现在需要126秒!

我可以做的任何优化使manager.dict()与python {}保持一致？

我对concurrent.futures有两个问题:

如何在python concurrent.futures中打破time.sleep()？

结论:time.sleep()不能被中断.一种解决方案是:您可以围绕它编写循环并进行短暂睡眠.

请参阅如何在python concurrent.futures中中断time.sleep()

concurrent.futures的个别超时？

结论:个人超时需要由用户实施.例如:对于每个超时,您可以调用wait().

请参阅 concurrent.futures的各个超时

题

是否ASYNCIO解决问题的论文？

我按照下面的代码在postgres数据库上实现并行选择查询:

https://tech.geoblink.com/2017/07/06/parallelizing-queries-in-postgresql-with-python/

我的基本问题是我有大约6k个需要执行的查询,我正在尝试优化这些选择查询的执行.最初它是一个where id in (...)包含所有6k谓词ID 的单个查询,但是我在运行它的机器上使用> 4GB的RAM遇到了查询问题,因此我决定将其拆分为6k个别查询,当同步保持时稳定的内存使用.然而,运行时间需要更长的时间,这对我的用例来说不是一个问题.即便如此,我也在尽量减少时间.

这就是我的代码:

class PostgresConnector(object):
    def __init__(self, db_url):
        self.db_url = db_url
        self.engine = self.init_connection()
        self.pool = self.init_pool()

    def init_pool(self):
        CPUS = multiprocessing.cpu_count()
        return multiprocessing.Pool(CPUS)

    def init_connection(self):
        LOGGER.info('Creating Postgres engine')
        return create_engine(self.db_url)

    def run_parallel_queries(self, queries):
        results = []
        try:
            for i in self.pool.imap_unordered(self.execute_parallel_query, queries):
                results.append(i)
        except Exception as exception:
            LOGGER.error('Error whilst executing %s queries in parallel: %s', len(queries), exception)
            raise
        finally:
            self.pool.close()
            self.pool.join()

        LOGGER.info('Parallel query ran producing %s sets of results of type: …

我正在编写代码以并行运行实验。我无法控制实验的内容，它们可能会公开使用subprocess.Popen或check_output运行一个或多个其他子进程。

我有两个条件：我希望能够终止超过超时的实验，并且希望终止于的实验KeyboardInterrupt。

大多数终止进程的方法都不能确保杀死所有子进程等。如果一个接一个地运行数百个实验，但是它们都生成了子进程，这些子进程在超时发生后仍然存活，并且据说该实验被终止，那么这显然是一个问题。

我现在处理该问题的方式包括添加代码以将实验配置存储在数据库中，生成从命令行加载和运行实验的代码，然后通过调用这些命令subprocess.Popen(cmd, shell=True, start_new_session=True)并使用os.killpg超时终止它们。

然后我的主要问题是：通过命令行调用这些实验很麻烦，因此有没有一种方法可以直接通过调用代码，multiprocessing.Process(target=fn)并在超时和？时获得与start_new_session=True+ 相同的效果。os.killpgKeyboardInterrupt

<file1>
def run_exp(config):
    do work
    return result

if __name__ == "__main__":
    save_exp(run_exp(load_config(sys.args)))

<file2>
def monitor(queue):
    active = set()  # active process ids
    while True:
        msg = queue.get()
        if msg == "sentinel":
             <loop over active ids and kill them with os.killpg>
        else:
            <add or remove id from active set>


def worker(args):
    id, queue = args …

我正在尝试使用 concurrent.futures 模块将一个需要很长时间的进程拆分为多个进程。附上下面的代码

主功能：

with concurrent.futures.ProcessPoolExecutor() as executor:
for idx, score in zip([idx for idx in range(dataframe.shape[0])],executor.map(get_max_fuzzy_score,[dataframe[idx:idx+1] for idx in range(dataframe.shape[0])])):
    print('processing '+str(idx+1)+' of '+str(dataframe.shape[0]+1))
    dataframe['max_row_score'].iloc[idx] = score

get_max_fuzzy_score 函数：

def get_max_fuzzy_score(picklepath_or_list, df):
import numpy as np
extracted_text_columns = list(df.filter(regex='extracted_text').columns)
data_list = [df[data].iloc[0] for data in extracted_text_columns if not df[data].isnull().values.any()]
try:
    size = len(picklepath_or_list)
    section_snippet_list = picklepath_or_list
except:
    section_snippet_list = pickle.load(open(picklepath_or_list,'rb'))
scores = []
for section_snippet in section_snippet_list:
    for data in data_list:
        scores.append(fuzz.partial_ratio(data,section_snippet))
score = max(scores)

return score

该函数采用几列的值，并从先前构建的列表中返回最大模糊分数。

这是我得到的错误： …

我正在尝试使用多处理填充一个numpy数组,在这篇文章之后.我在Mac上运行得很好,但是当我将它移植到Ubuntu时,我会在很多时候遇到分段错误.

我已经将代码简化为以下最小的示例:

import numpy as np
from multiprocessing import sharedctypes

a = np.ctypeslib.as_ctypes(np.zeros((224,224,3)))
print("Made a, now making b")
b = sharedctypes.RawArray(a._type_, a)
print("Finished.")

在Ubuntu 16.04上,使用Python 3.6.5和numpy 1.15.4(与我的Mac上相同的版本),我得到了输出

Made a, now making b
Segmentation fault (core dumped)

现在,我可以稍微更改数组维度,在某些情况下它可以工作(例如,将第一个224更改为100并且它可以工作).但主要是它会出现故障.

有人可以提供任何见解吗？

我看到一篇关于2016年相关主题的帖子没有人回复,另一篇涉及我没有使用的指针.

PS-我是指定a为多维数组还是扁平数组(例如np.zeros(224*224*3))似乎没有任何区别.如果我更改数据类型(例如float to int),它似乎也没有区别; 它失败了.

还有一个更新:即使在原始帖子的代码中设置"size = 224",也会导致两个不同版本的numpy的Ubuntu机器出现seg故障,但在Mac上工作正常.

因此，假设您有一个 Python 进程，它正在以每秒大约 500 行的速度实时收集来自排队系统的数据（这可以进一步并行化以减少到大约 50 ps）并将其附加到DataFrame：

rq = MyRedisQueue(..)
df = pd.DataFrame()
while 1:
    recv = rq.get(block=True)
    # some converting
    df.append(recv, ignore_index = True)

现在的问题是：如何根据这些数据利用 CPU？所以我充分认识到的局限性GIL，看着成多处理管理器 的命名空间，这里太，但它看起来像有一些缺点关于延迟的centerally保持数据帧。在深入研究之前，我还尝试pool.map了我认为pickle在进程之间应用的方法，这种方法很慢并且开销太大。

所以在所有这一切之后，我终于想知道，如何（如果）每秒 500 行（甚至每秒 50 行）的插入可以传输到不同的进程，同时还有一些 CPU 时间用于对子项中的数据应用统计和启发式流程？

也许在两个进程之间实现自定义 tcp 套接字或排队系统会更好？或者在pandas其他库中是否有一些实现可以真正允许快速访问父进程中的一个大数据框？我爱熊猫！

• 我正在使用 amultiprocessing.Pool在 1 个或多个子进程中调用函数来生成大量数据。
• 工作进程创建一个multiprocessing.shared_memory.SharedMemory对象并使用由 分配的默认名称shared_memory。
• worker 将SharedMemory对象的字符串名称返回给主进程。
• 在主进程中，SharedMemory对象被链接、消费，然后取消链接和关闭。

在关闭时，我看到来自resource_tracker以下方面的警告：

/usr/local/lib/python3.8/multiprocessing/resource_tracker.py:216: UserWarning: resource_tracker: There appear to be 10 leaked shared_memory objects to clean up at shutdown
  warnings.warn('resource_tracker: There appear to be %d '
/usr/local/lib/python3.8/multiprocessing/resource_tracker.py:229: UserWarning: resource_tracker: '/psm_e27e5f9e': [Errno 2] No such file or directory: '/psm_e27e5f9e'
  warnings.warn('resource_tracker: %r: %s' % (name, e))
/usr/local/lib/python3.8/multiprocessing/resource_tracker.py:229: UserWarning: resource_tracker: '/psm_2cf099ac': [Errno 2] No such file or directory: '/psm_2cf099ac'
<8 more similar …

语境：

• 一个使用 aconcurrent.futures.process.ProcessPool来执行代码的Python 应用服务器
• 我们有时想在不重启整个服务器进程的情况下热重载导入的代码

（是的，我知道importlib.reload有警告）

为了让它工作，我想我必须在进程池管理的importlib.reload每个multiprocessing进程中执行。

有没有办法向进程池中的所有进程提交一些东西？