Process evidences in parallel

Author: mkardas

helpers.py

@@ -5,6 +5,7 @@
 from elasticsearch_dsl import connections
 from tqdm import tqdm
 import pandas as pd
+from joblib import delayed, Parallel
 
 class Helper:
     def split_pc_pickle(self, path, outdir="pc-parts", parts=8):
@@ -16,18 +17,22 @@ def split_pc_pickle(self, path, outdir="pc-parts", parts=8):
             part = PaperCollection(pc[i:i + step])
             part.to_pickle(outdir / f"pc-part-{idx:02}.pkl")
 
-    def evidences_for_pc(self, path):
+    def _evidences_for_pc(self, path):
         path = Path(path)
         pc = PaperCollection.from_pickle(path)
         cell_evidences = CellEvidenceExtractor()
         connections.create_connection(hosts=['10.0.1.145'], timeout=20)
         raw_evidences = []
         for paper in tqdm(pc):
-            raw_evidences.append(cell_evidences(paper, paper.tables))
+            raw_evidences.append(cell_evidences(paper, paper.tables, paper_limit=100, corpus_limit=20))
         raw_evidences = pd.concat(raw_evidences)
         path = path.with_suffix(".evidences.pkl")
         raw_evidences.to_pickle(path)
 
+    def evidences_for_pc(self, pattern="pc-parts/pc-part-??.pkl", jobs=-1):
+        pickles = sorted(Path(".").glob(pattern))
+        Parallel(backend="multiprocessing", n_jobs=jobs)(delayed(self._evidences_for_pc)(path) for path in pickles)
+
     def merge_evidences(self, output="evidences.pkl", pattern="pc-parts/pc-part-*.evidences.pkl"):
         pickles = sorted(Path(".").glob(pattern))
         evidences = [pd.read_pickle(pickle) for pickle in pickles]