加快多个CSV文件的加载速度

问题描述 投票：0回答：1

我正在尝试为法国公共药物数据库编写一个解析器/API（https://base-donnees-publique.medicaments.gouv.fr/）。它由八个 CSV 文件（实际上是 TSV，因为他们使用制表符）组成，每个文件从几 kB 到 4 MB，最大的有约 20000 行（每行代表一种药物及其名称、代码、价格等）。

由于这些文件可能会定期出现，我想直接解析它们，而不是创建一个更干净的数据库（因为我可能必须定期重新创建它）。

导入这些文件需要一点时间（大约一秒钟），所以我尝试加快一点速度并对几种方法进行了一些基准测试，我惊讶地发现最基本的方法似乎也是最快的。

这是我的测试代码（抱歉它很长）。每个文件都与一个专用类关联来解析其行。基本上，这些类是命名元组，具有自定义类方法来解析日期、数字等。

import pathlib
import enum
import datetime
from decimal import Decimal
from collections import namedtuple
import csv


def parse_date(date: str) -> datetime.datetime:
    return datetime.datetime.strptime(date, "%d/%m/%Y").date()


def parse_date_bis(date: str) -> datetime.datetime:
    return datetime.datetime.strptime(date, "%Y%m%d").date()


def parse_text(text):
    if not text:
        return ""
    return text.replace("<br>", "\n").strip()


def parse_list(raw):
    return raw.split(";")


def parse_price(price: str) -> Decimal:
    if not price:
        return None
    # Handles cases like "4,417,08".
    price = '.'.join(price.rsplit(",", 1)).replace(',', '')
    return Decimal(price)


def parse_percentage(raw: str) -> int:
    if not raw:
        return None
    return int(raw.replace("%", "").strip())


class StatutAdministratifPresentation(enum.Enum):
    ACTIVE = "Présentation active"
    ABROGEE = "Présentation abrogée"


class EtatCommercialisation(enum.Enum):
    DC = "Déclaration de commercialisation"
    S = "Déclaration de suspension de commercialisation"
    DAC = "Déclaration d'arrêt de commercialisation"
    AC = "Arrêt de commercialisation (le médicament n'a plus d'autorisation)"


class MotifAvisSMR(enum.Enum):
    INSCRIPTION = "Inscription (CT)"
    RENOUVELLEMENT = "Renouvellement d'inscription (CT)"
    EXT = "Extension d'indication"
    EXTNS = "Extension d'indication non sollicitée"
    REEV_SMR = "Réévaluation SMR"
    REEV_ASMR = "Réévaluation ASMR"
    REEV_SMR_ASMR = "Réévaluation SMR et ASMR"
    REEV_ETUDE = "Réévaluation suite à résultats étude post-inscript"
    REEV_SAISINE = "Réévaluation suite saisine Ministères (CT)"
    NOUV_EXAM = "Nouvel examen suite au dépôt de nouvelles données"
    MODIF_COND = "Modification des conditions d'inscription (CT)"
    AUTRE = "Autre demande"


class ImportanceSMR(enum.Enum):
    IMPORTANT = "Important"
    MODERE = "Modéré"
    FAIBLE = "Faible"
    INSUFFISANT = "Insuffisant"
    COMMENTAIRES = "Commentaires"
    NP = "Non précisé"


class ImportanceASMR(enum.Enum):
    COM = "Commentaires sans chiffrage de l'ASMR"
    I = "I"
    II = "II"
    III = "III"
    IV = "IV"
    V = "V"
    NP = "Non précisée"
    SO = "Sans objet"


class Specialite(namedtuple("Specialite", ("cis", "denomation", "forme", "voies_administration", "statut_amm", "type_amm", "commercialisation", "date_amm", "statut_bdm", "numero_autorisation_europeenne", "titulaire", "surveillance_renforcee"))):
    @classmethod
    def from_line(cls, line):
        line[2] = line[2].replace("  ", " ").strip()
        line[3] = parse_list(line[3])
        line[7] = parse_date(line[7])
        line[10] = line[10].strip()  # There are often leading spaces here (like ' OPELLA HEALTHCARE FRANCE').
        return cls(*line)


class Presentation(namedtuple("Specialite", ("cis", "cip7", "libelle", "statut", "commercialisation", "date_commercialisation", "cip13", "agrement_collectivites", "taux_remboursement", "prix", "prix_hors_honoraires", "montant_honoraires", "indications_remboursement"))):
    @classmethod
    def from_line(cls, line):
        if line[3] == "Présentation active":
            line[3] = StatutAdministratifPresentation.ACTIVE
        else:
            line[3] = StatutAdministratifPresentation.ABROGEE
        line[4] = {
            "Déclaration de commercialisation": EtatCommercialisation.DC,
            "Déclaration de suspension de commercialisation": EtatCommercialisation.S,
            "Déclaration d'arrêt de commercialisation": EtatCommercialisation.DAC,
            "Arrêt de commercialisation (le médicament n'a plus d'autorisation)": EtatCommercialisation.AC
        }.get(line[4])
        line[5] = parse_date(line[5])
        line[7] = True if line[7] == "oui" else False
        line[8] = parse_percentage(line[8])
        line[9] = parse_price(line[9])
        line[10] = parse_price(line[10])
        line[11] = parse_price(line[11])
        line[12] = parse_text(line[12])
        return cls(*line)


class Composition(namedtuple("Composition", ("cis", "element", "code", "substance", "dosage", "ref_dosage", "nature_composant", "cle"))):
    @classmethod
    def from_line(cls, line):
        line.pop(-1)
        return cls(*line)


class AvisSMR(namedtuple("AvisSMR", ("cis", "dossier_has", "motif", "date", "valeur", "libelle"))):
    @classmethod
    def from_line(cls, line):
        line[2] = MotifAvisSMR(line[2])
        line[3] = parse_date_bis(line[3])
        line[4] = ImportanceSMR(line[4])
        line[5] = parse_text(line[5])
        return cls(*line)


class AvisASMR(namedtuple("AvisASMR", ("cis", "dossier_has", "motif", "date", "valeur", "libelle"))):
    @classmethod
    def from_line(cls, line):
        line[2] = MotifAvisSMR(line[2])
        line[3] = parse_date_bis(line[3])
        line[4] = ImportanceASMR(line[4])
        line[5] = parse_text(line[5])
        return cls(*line)


class AvisCT(namedtuple("AvisCT", ("dossier_has", "lien"))):
    @classmethod
    def from_line(cls, line):
        return cls(*line)


FILE_MATCHES = {
    "CIS_bdpm.txt": Specialite,
    "CIS_CIP_bdpm.txt": Presentation,
    "CIS_COMPO_bdpm.txt": Composition,
    "CIS_HAS_ASMR_bdpm.txt": AvisASMR,
    "CIS_HAS_SMR_bdpm.txt": AvisSMR,
    "HAS_LiensPageCT_bdpm.txt": AvisCT
}


def sequential_import_file_data(filename, cls):
    result = {cls: []}
    with (pathlib.Path("data") / filename).open("r", encoding="latin1") as f:
        rows = csv.reader(f, delimiter="\t")
        for line in rows:
            data = cls.from_line(line)
            result[cls].append(data)
    return result


def import_data_sequential():
    results = []
    for filename, cls in FILE_MATCHES.items():
        results.append(sequential_import_file_data(filename, cls))


from multiprocessing.pool import ThreadPool

def import_data_mp_tp(n=2):
    pool = ThreadPool(n)
    results = []
    for filename, cls in FILE_MATCHES.items():
        results.append(pool.apply_async(
            sequential_import_file_data,
            (filename, cls)
        ))
    results = [r.get() for r in results]


from multiprocessing.pool import Pool

def import_data_mp_p(n=2):
    pool = Pool(n)
    results = []
    for filename, cls in FILE_MATCHES.items():
        results.append(pool.apply_async(
            sequential_import_file_data,
            (filename, cls)
        ))
    results = [r.get() for r in results]


import asyncio
import aiofiles
from aiocsv import AsyncReader

async def async_import_file_data(filename, cls):
    results = {cls: []}
    async with aiofiles.open(
        (pathlib.Path("data") / filename),
        mode="r",
        encoding="latin1"
    ) as afp:
        async for line in AsyncReader(afp, delimiter="\t"):
            data = cls.from_line(line)
            results[cls].append(data)
    return results


def import_data_async():
    results = []
    for filename, cls in FILE_MATCHES.items():
        results.append(asyncio.run(async_import_file_data(filename, cls)))


def main():
    import timeit
    print(
        "Sequential:",
        timeit.timeit(lambda: import_data_sequential(), number=10)
    )
    print(
        "Multi ThreadPool:",
        timeit.timeit(lambda: import_data_mp_tp(), number=10)
    )
    print(
        "Multi Pool:",
        timeit.timeit(lambda: import_data_mp_p(), number=10)
    )
    print(
        "Async:",
        timeit.timeit(lambda: import_data_async(), number=10)
    )


if __name__ == "__main__":
    main()

所以当我运行它时，我得到以下结果。

Sequential: 9.821639589001279
Multi ThreadPool: 10.137484730999859
Multi Pool: 12.531487682997977
Async: 30.953154197999538

迭代所有文件及其所有行的最基本解决方案似乎也是最快的。

那么，我是否做错了什么会减慢导入速度？或者说有这样的时差是正常/预期的吗？

编辑2023-08-15：我意识到，由于我需要解析的所有文件都是TSV（并且它们的值中不包含制表符），我仍然可以通过使用简单的

line.strip('\n').split('\t')
来加快解析速度而不是 CSV 模块，运行时间又节省了 40%。 :) 当我有这个数据库的完整 API 时，我可能会发布一个要点。

python performance csv asynchronous
1个回答
3
投票

像往常一样：在代码上运行分析器以查看其时间花在哪里。 （这是 PyCharm 的，它包装了 stdlib 

cProfile
。）

顺序：7.865187874995172

嗯，好吧。 

strptime
，我知道会被
datetime.datetime.strptime
打电话。另外，奇怪的是，
getlocale
...为什么我们需要那里的语言环境？单击调用图显示 
strptime
实际上查找当前区域设置，并且有一堆锁等等 - 如果我们用我们自己的实现替换这些 
parse_date
会怎么样？

def parse_date(date: str) -> datetime.date:
    d, m, y = (int(x) for x in date.split("/", 2))
    return datetime.date(2000 + y, m, d)


def parse_date_bis(date: str) -> datetime.datetime:
    y = int(date[:4])
    m = int(date[4:6])
    d = int(date[6:8])
    return datetime.datetime(y, m, d)

顺序：3.8978060420195106

好的，我们做饭了！ 52% 的改进就在那里！

（它没有出现在此处的屏幕截图中，因为我是一只愚蠢的鹅裁剪它，但是 

re
在引擎盖下使用的 
strptime
东西也立即掉落了。）

现在让我们假设在那些热门 

@lru_cache(maxsize=None)
函数上会有很多相同的日期和 slap 
parse_date_*
（RAM 灵活，无界缓存），运行代码并打印出缓存信息:

Sequential: 3.2240814580000006
CacheInfo(hits=358989, misses=6991, maxsize=None, currsize=6991)
CacheInfo(hits=221607, misses=513, maxsize=None, currsize=513)

我觉得还不错，我们最后一个号码还有 15% 的折扣。

不过，

parse_price
显然也可以使用缓存：

Sequential: 2.928746833000332
CacheInfo(hits=358989, misses=6991, maxsize=None, currsize=6991)
CacheInfo(hits=221607, misses=513, maxsize=None, currsize=513)
CacheInfo(hits=622064, misses=4096, maxsize=None, currsize=4096)

嘿，谁知道，数据中只有 4096 个单独的价格字符串。

如果你有足够的内存，剩下的解析函数也可以使用缓存，但是通过一点点分析和解析苦劳，它现在的速度提高了 2.7 倍 [当运行所有内容 10 次时，这意味着这些缓存将会很热 – 单个run 的加速并不那么显着]，不需要并行处理。魔法！

为了让比赛场地更加均匀，这里有一个 

hyperfine
基准测试，其中 Python 解释器对于每次导入都从头开始（每个解释器仅运行一次导入）：

$ hyperfine 'python3 so76781391-orig.py' 'python3 so76781391-opt.py' --warmup 5 --min-benchmarking-time 10
Benchmark 1: python3 so76781391-orig.py
  Time (mean ± σ):     363.0 ms ±   2.7 ms    [User: 340.8 ms, System: 20.7 ms]
  Range (min … max):   358.9 ms … 367.9 ms    27 runs

Benchmark 2: python3 so76781391-opt.py
  Time (mean ± σ):     234.1 ms ±   2.5 ms    [User: 215.6 ms, System: 17.0 ms]
  Range (min … max):   228.2 ms … 238.5 ms    42 runs

Summary
  'python3 so76781391-opt.py' ran
    1.55 ± 0.02 times faster than 'python3 so76781391-orig.py'

所以，通过快速查看分析器，具体速度提升了 55%（以及一些额外的优化，例如不在 

from_line
函数中创建映射字典等）。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.