1

我有一个数据帧，其中包含大约 15k 路径的音频文件，我想在其上执行操作（人为添加噪声）。一般来说，整个事情都有效，但即使记录较少（16）条记录也需要很长时间。问题不是函数的执行时间，而是所有初始化之前的时间。

start = time.time()
data_augmented = data_augmented.swifter.progress_bar(True, desc="Merge Sounds") \
        .apply(merge_sounds(**settings), axis=1)
print(f"{time.time-(start)} - Map Timer")

Merge Sounds: 100%|█████████████████████████████| 16/16 [00:07<00:00,  2.09it/s]
26.973325729370117 - Map Timer

正如您在此处看到的，初始化所需的时间几乎是 Lambda 函数 ( merge_sounds) 运行时间的 4 倍。在这种情况下，initialization-time我指的是elapsed_time_measured_by_myself - elapsed_time_measured_by_tqdm26.97.. - 7 = 19.97

    start = time.time()
    lambda_fn = merge_sounds(**settings) #doesnt work if i put it in the line below.
    data_augmented = process_map(lambda_fn, data_augmented, max_workers=threads,
                                     desc=f"Merge_sounds [{threads} Threads]")
    print(f"{time.time-(start)} - Map Timer")

卡在：

Merge_sounds [16 Threads]:   0%|                         | 0/16 [00:00<?, ?it/s]

    with Pool(processes=16) as pool:
        data_augmented = pool.map(merge_sounds(**settings), tqdm(data_augmented, desc=f"Merge Sounds: {16} Threads"))

卡在：

Merge Sounds: 16 Threads:  38%|██████          | 6/16 [00:00<00:00, 4697.75it/s]

我知道并行化对于较小的数据集没有意义，我只是很困惑为什么我可以轻松地将代码中的所有内容并行化，而我只是无法在这里取得成功。后来我在大量数据上运行这段代码，所以如果并行性可行，我会很高兴。

Map 中使用的函数是：

def merge_sounds(**settings):
    _range = settings.get("snr_range", (0.15, 0.65))
    assert len(_range), "snr_range -> e.g. (0.15, 0.75)"
    target_sample_rate = settings.get("target_sample_rate", "16000")

    if "target_path" not in settings.keys():
        raise Exception("please Specify target_path in Settings-Dict")
    target_path = Path(settings["target_path"])
    target_path.mkdir(parents=True, exist_ok=True)

    def __call__(item):
        _target_path = item["path_augmented"]

        snr = round(uniform(_range[0], _range[1]), 4)
        pad_idx = item.name

        yp, _ = IO.load(item["path"], sample_rate=target_sample_rate)
        yn, _ = IO.load(item["path_noise"], sample_rate=target_sample_rate)
        item["snr"] = snr

        y_augmented = Effect.add_noise(yp, yn, snr=snr, pad_idx=pad_idx)
        IO.save_wav(y_augmented, _target_path, target_sample_rate)
        return item

    return __call__

有什么我忘记并行化映射函数的吗（似乎在我的代码中的其他任何地方都可以使用这种变体之一，就像预期的那样）

Ty 进阶。

4

1 回答 1

0

如果通过绕过 TQDM 解决了此问题。我压缩了我需要这样的东西的列

    _paths_in = _df["path_input"]
    _paths_out = _df["path_output"]
    _path_noise = _df["path_noise"]
    job = zip(_paths_in, _path_noise, _paths_out, _filter_jobs)

然后我只是将它传递给 MultiProcessor 函数。

    jobs = list(enumerate((zip_jobs(df))))

    with Pool(processes=_threads) as pool:
        data_augmented = pool.map(execute_job, tqdm(jobs, desc=f"Audio-Augmentation: {_threads} Threads"))

merge_sounds和新execute_job的差不多，只是改了，函数需要什么参数。

于 2021-05-05T12:39:03.397 回答