Use 'Distributed.pmap' in 'ParallelTableTransform'

Author: eliascarv

Project.toml

@@ -9,6 +9,7 @@ CategoricalArrays = "324d7699-5711-5eae-9e2f-1d82baa6b597"
 CoDa = "5900dafe-f573-5c72-b367-76665857777b"
 ColumnSelectors = "9cc86067-7e36-4c61-b350-1ac9833d277f"
 DataScienceTraits = "6cb2f572-2d2b-4ba6-bdb3-e710fa044d6c"
+Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"
 Distributions = "31c24e10-a181-5473-b8eb-7969acd0382f"
 InverseFunctions = "3587e190-3f89-42d0-90ee-14403ec27112"
 LinearAlgebra = "37e2e46d-f89d-539d-b4ee-838fcccc9c8e"
@@ -28,6 +29,7 @@ CategoricalArrays = "0.10"
 CoDa = "1.2"
 ColumnSelectors = "0.1"
 DataScienceTraits = "0.3"
+Distributed = "1.9"
 Distributions = "0.25"
 InverseFunctions = "0.1"
 LinearAlgebra = "1.9"

src/TableTransforms.jl

@@ -24,6 +24,7 @@ using Distributions: ContinuousUnivariateDistribution, Normal
 using InverseFunctions: NoInverse, inverse as invfun
 using StatsBase: AbstractWeights, Weights, sample
 using Transducers: tcollect
+using Distributed: pmap
 using NelderMead: optimise
 
 import Distributions: quantile, cdf

src/transforms/parallel.jl

@@ -41,8 +41,7 @@ isrevertible(p::ParallelTableTransform) = any(isrevertible, p.transforms)
 
 function apply(p::ParallelTableTransform, table)
   # apply transforms in parallel
-  f(transform) = apply(transform, table)
-  vals = tcollect(f(t) for t in p.transforms)
+  vals = pmap(t -> apply(t, table), p.transforms)
 
   # retrieve tables and caches
   tables = first.(vals)
@@ -122,9 +121,7 @@ function reapply(p::ParallelTableTransform, table, cache)
   caches = cache[1]
 
   # reapply transforms in parallel
-  f(t, c) = reapply(t, table, c)
-  itr = zip(p.transforms, caches)
-  tables = tcollect(f(t, c) for (t, c) in itr)
+  tables = pmap((t, c) -> reapply(t, table, c), p.transforms, caches)
 
   # features and metadata
   splits = divide.(tables)