Mlj.jl: Help with loading code on multiple processes for paralleled tuning of a pipeline
Hi,
I am using @pipeline to construct a MLJ pipeline with my custom pre-processor. This is a classifier problem and I used LDA from MultivariateStats. Basically, this is my custom transformer and its fit/predict looks like.
@with_kw_noshow mutable struct MyCustomTransformer <:MLJBase.Unsupervised
skew::Float64= 0.0
t1pad::Int64 = 1
pow::Int64 = 1
end
# I want to transform all features
function MLJBase.fit(transformer::MyCustomTransformer, verbosity::Int, X)
fitresult = collect(Tables.schema(X).names)
report = NamedTuple()
return fitresult, nothing, report
end
MLJBase.fitted_params(::MyCustomTransformer, fitresult) = (features_to_keep=fitresult,)
function MLJBase.transform(transformer::MyCustomTransformer, fitresult, X)
# row-wise transformation
spectrum = map(
row -> fid_to_features(
row.fid,
t1pad=transformer.t1pad,
win=t->sinebell(t, skew=transformer.skew, pow=transformer.pow)
),
eachrow(X)
)
# Put array of array result in DataFrame and unpack the inner array
result_df = DataFrame([tuple(x...) for x in spectrum])
@info typeof(result_df) size(result_df)
result_df
end
And here is how I wrap it
# Partition data (train+validation)/test by 80/20
train, test = partition(eachindex(y), 0.8, shuffle=true)
# Then, I can tune the 'hyper-parameter' skew by
classifier = @pipeline MyPipe(preprocessor= MyCustomTransformer(), model = lda_model) prediction_type=:probabilistic
#Then wrap it with a TunedModel
ranges = [range(classifier, :(preprocessor.t1pad), lower=1, upper=2, scale = :linear),
range(classifier, :(preprocessor.skew), lower=0.1, upper=0.3, scale = :linear),
range(classifier, :(preprocessor.pow), lower=1, upper=2, scale = :linear)]
tuned_model = TunedModel(; model=classifier,
tuning = Grid(resolution=2),
resampling = CV(nfolds=5, shuffle=true),
measure = accuracy,
ranges = ranges,
operation = predict_mode)
# Train, test, get best hyper-params (skew, pow, etc)
mach = machine(tuned_model, X, y)
fit!(mach, rows=train)
I have set this to use multi processes by setting MLJ.default_resource(ComputationalResources.CPUProcesses()). However, this still runs very slow considering the data is small (only has ~60 rows). The initial data is a DataFrame of AxisArray (:fid column) and transformed to features/spectrum with some pre-processing (essentially array calculation). Depending on the t1pad hyper-parameter, this could result in ~1000-4000 columns.
WIth that resolution=2, this results in 8 models to be evaluated. I am just wondering why is that slow? I lodged a job in HPC yesterday and still not finished. ~18hrs. I expected this to be not that slow since I uses a simple model.
I was implementing this grid search manually over my pre-processing function with nested loops and it's not this slow. Is there any obvious mistake I am having here? Is the data X should be put in the pipeline instead to optimize this, but how can I build the TunedModel to tune my hyper-parameters if I fetch the data in the pipeline?
Also, what is the best way to identify this bottleneck? Is there any monitoring system in Pipeline/TunedModel instead of having a 'black-box'? I noticed there is a progress bar but this doesn't show when I use multi-processes.
I use newest release of MLJ (v0.8.0) and Julia 1.3.1.
Thanks!
darrencl
All 12 comments
@darrencl what are the specs of the HPC?.
Also there are a lot of things to consider when doing parallel computing (data race, memory requirements etc.).
I would suggest using the MLJ.default_resource(ComputationalResources.CPU1()) instead and seeing the effect of resetting the number of threads used by BLAS with the code using LinearAlgebra; BLAS.set_num_threads(n) (where n is the number of threads to be used).
Also when u used MLJ.default_resource(ComputationalResources.CPUProcesses())did u also do using Distributed; addprocs(n) (where n is the no. of physical cores to be used) to increase the number of cores used in the computation.
@ablaom @tlienart any additions?
OkonSamuel
on 15 Feb 2020
@darrencl Thanks for reporting the speed issues.
Had a quick look over the code and I don't see anything wrong with what you are doing. It's a bit hard to investigate without a more complete picture (concrete data).
In terms of getting more reporting from the pipeline, you could (in serial mode) increase the verbosity level, eg. fit!(mach, verbosity=4). What if you just train a MyPipe instance without the tuning wrapper. Is this slow?
ablaom
on 17 Feb 2020
The high dimensionality of your problem is likely a cause of issues (p >>> n) i.e. LDA itself is likely super slow. Consider either doing logistic regression or subspace LDA, this quote is from multivariatestats:
#### SubspaceLDA
# When the dimensionality is much higher than the number of samples,
# it makes more sense to perform LDA on the space spanned by the
# within-group scatter.
@OkonSamuel wrote an interface for it too so it should be a mere case of swapping the model name to see if that might bring a significant speedup
tlienart
on 17 Feb 2020
@OkonSamuel @ablaom @tlienart Thanks a lot for the response and suggestions guys! I'll try those.
I haven't use this using Distributed; addprocs(n). Thought it automatically use the available CPUs when I specify CPUProcesses() as default resource.
This HPC has several worker nodes, each with quite old CPU - AMD Opteron(tm) Processor 6278 and ~264GB of memory (not sure the spec of the memory, but pretty sure it's quite old too).
I think without the tuning wrapper it's still slow. Took ~2.5 hours to finish the pipeline. However, I notice the second time I train this (loop through dataset1 and dataset2, but I still re-initialize my pipeline in the loop), it only took 4 seconds, what's causing this?
Just trying to understand the logging
[ Info: Training NodalMachine{MyCustomTransformer} @ 3…30.
Does 3...30 mean anything anyway?
Then it makes sense if LDA is pretty slow in this case, I'll try the SubspaceLDA and logistic regression instead then.
darrencl
on 19 Feb 2020
For tips on parallel computing in Julia I recommend https://github.com/mbauman/ParallelWorkshop2019
The least obvious thing to know is that you generally want to load your project environment on every process.
ablaom
on 19 Feb 2020
@OkonSamuel @ablaom @tlienart Finally found where the bottleneck is. With the logging I found it very slow on training the MyCustomTransformer and investigate from there. It seems that unpacking to DataFrame this way is very slow in high-dimensional data.
DataFrame([tuple(x...) for x in spectrum])
This can be replaced by
DataFrame!([getindex.(spectrum, i) for i in 1:maximum(length.(spectrum))]);
and this significantly improves the speed.
However, I'm still not sure as to why the second training time is very fast despite re-initializing the objects.
I think what would be great is to have a logging (when user specify high verbosity or maybe another option) on elapsed time on each training stage (i.e. fit and transform).
Anyway, I will close this issue for now as the problem is solved and the bottleneck is nothing to do with MLJ.jl.
darrencl
on 20 Feb 2020
You could also just use a julia native table such as a "row table" or "column table". (You can give MLJ models any Tables.jl compatible table). A row table is any iterable (eg vector) of identically typed named tuples. A column table is a named tuple of identically-lengthed vectors. You can construct one by hand or from any other compatible table. EG:
julia> using MLJ, Tables, RDatasets
julia> A = rand("ABC", 2, 3)
2×3 Array{Char,2}:
'C' 'B' 'C'
'C' 'C' 'C'
julia> matrixtable = Tables.table(A)
Tables.MatrixTable{Array{Char,2}}(Symbol[:Column1, :Column2, :Column3], Dict(:Column2 => 2,:Column1 => 1,:Column3 => 3), ['C' 'B' 'C'; 'C' 'C' 'C'])
julia> rowtable = Tables.rowtable(matrixtable)
2-element Array{NamedTuple{(:Column1, :Column2, :Column3),Tuple{Char,Char,Char}},1}:
(Column1 = 'C', Column2 = 'B', Column3 = 'C')
(Column1 = 'C', Column2 = 'C', Column3 = 'C')
julia> dataframe = dataset("datasets", "iris")
150×5 DataFrame
│ Row │ SepalLength │ SepalWidth │ PetalLength │ PetalWidth │ Species │
│ │ Float64 │ Float64 │ Float64 │ Float64 │ Categorical… │
├─────┼─────────────┼────────────┼─────────────┼────────────┼──────────────┤
│ 1 │ 5.1 │ 3.5 │ 1.4 │ 0.2 │ setosa │
│ 2 │ 4.9 │ 3.0 │ 1.4 │ 0.2 │ setosa │
│ 3 │ 4.7 │ 3.2 │ 1.3 │ 0.2 │ setosa │
│ 4 │ 4.6 │ 3.1 │ 1.5 │ 0.2 │ setosa │
│ 5 │ 5.0 │ 3.6 │ 1.4 │ 0.2 │ setosa │
│ 6 │ 5.4 │ 3.9 │ 1.7 │ 0.4 │ setosa │
│ 7 │ 4.6 │ 3.4 │ 1.4 │ 0.3 │ setosa │
⋮
│ 143 │ 5.8 │ 2.7 │ 5.1 │ 1.9 │ virginica │
│ 144 │ 6.8 │ 3.2 │ 5.9 │ 2.3 │ virginica │
│ 145 │ 6.7 │ 3.3 │ 5.7 │ 2.5 │ virginica │
│ 146 │ 6.7 │ 3.0 │ 5.2 │ 2.3 │ virginica │
│ 147 │ 6.3 │ 2.5 │ 5.0 │ 1.9 │ virginica │
│ 148 │ 6.5 │ 3.0 │ 5.2 │ 2.0 │ virginica │
│ 149 │ 6.2 │ 3.4 │ 5.4 │ 2.3 │ virginica │
│ 150 │ 5.9 │ 3.0 │ 5.1 │ 1.8 │ virginica │
julia> rowtable2 = Tables.rowtable(dataframe);
julia> rowtable2 |> pretty
┌─────────────┬────────────┬─────────────┬────────────┬─────────────────────── ⋯
│ SepalLength │ SepalWidth │ PetalLength │ PetalWidth │ Species ⋯
│ Float64 │ Float64 │ Float64 │ Float64 │ CategoricalString{UInt ⋯
│ Continuous │ Continuous │ Continuous │ Continuous │ Multiclass{3} ⋯
├─────────────┼────────────┼─────────────┼────────────┼─────────────────────── ⋯
│ 5.1 │ 3.5 │ 1.4 │ 0.2 │ setosa ⋯
│ 4.9 │ 3.0 │ 1.4 │ 0.2 │ setosa ⋯
│ 4.7 │ 3.2 │ 1.3 │ 0.2 │ setosa ⋯
│ 4.6 │ 3.1 │ 1.5 │ 0.2 │ setosa ⋯
│ 5.0 │ 3.6 │ 1.4 │ 0.2 │ setosa ⋯
│ 5.4 │ 3.9 │ 1.7 │ 0.4 │ setosa ⋯
│ 4.6 │ 3.4 │ 1.4 │ 0.3 │ setosa ⋯
│ 5.0 │ 3.4 │ 1.5 │ 0.2 │ setosa ⋯
│ 4.4 │ 2.9 │ 1.4 │ 0.2 │ setosa ⋯
│ 4.9 │ 3.1 │ 1.5 │ 0.1 │ setosa ⋯
│ 5.4 │ 3.7 │ 1.5 │ 0.2 │ setosa ⋯
│ 4.8 │ 3.4 │ 1.6 │ 0.2 │ setosa ⋯
│ 4.8 │ 3.0 │ 1.4 │ 0.1 │ setosa ⋯
│ 4.3 │ 3.0 │ 1.1 │ 0.1 │ setosa ⋯
│ ⋮ │ ⋮ │ ⋮ │ ⋮ │ ⋮ ⋯
└─────────────┴────────────┴─────────────┴────────────┴─────────────────────── ⋯
Glad you found the progressive loggin helpful. BTW I have now implemented progress meters for evaluate! for distributed and multithreading, which should help.
ablaom
on 20 Feb 2020
@ablaom @OkonSamuel Hi guys, I am trying to use the multiprocessing and multithreading to accelerate TunedModel with this following options acceleration=ComputationalResources.CPUProcesses() and acceleration_resampling=ComputationalResources.CPUThreads(). The threads seems to work fine without multiprocessing, so no issue on that.
Also, I added these codes to add processes and threads as suggested above.
# Load project environment in all process as suggested in the workshop materials
using Distributed
addprocs(32, exeflags=`--project=$@__DIR__`)
# Add threads
using LinearAlgebra
BLAS.set_num_threads(32)
With that, I have this error:
ERROR: LoadError: On worker 2:
KeyError: key MLJTuning [03970b2e-30c4-11ea-3135-d1576263f10f] not found
which seems can be addressed by adding @everywhere using MLJTuning. However, with that I am experiencing the same issue, but this time with my pipeline MyPipe and keeps on and on in other stuffs. My question is, should I use @everywhere in all of my definitions? What is the best way to do this?
Thanks!
darrencl
on 26 Feb 2020
@darrencl when using multiple processes every code (model structs, pipelines and functions) defined by you has to be defined using @everywhere macro. You also have to load MLJ and any other external package you wish to use (e.g Plots.jl) using the @everywhere macro. Here is a quick example.
#load Distributed module and add specified number of processors.
using Distributed
addprocs(32, exeflags=`--project=$@__DIR__`)
#load MLJ and Plots
@everywhere using MLJ
@everywhere using Plots #MLJ and any other package you wish to use for distributed computing must be loaded using the @everywhere macro.
#Define data
X = (age = [23, 45, 34, 25, 67],
gender = categorical(['m', 'm', 'f', 'm', 'f']));
height = [67.0, 81.5, 55.6, 90.0, 61.1]; #No need to define data or machine everywhere
#Make pipeline model type visible to all worker processes
@everywhere @load KNNRegressor # loaded KNNRegressor model every to make the model type visible to every process. (Always remember to do this when using Models from external packages.)
@everywhere @pipeline MyPipe(X -> coerce(X, :age=>Continuous),
hot = OneHotEncoder(),
regressor = KNNRegressor(K=3),
target =UnivariateStandardizer() ) # pipelines and every function or type introduced by you have to be defined using the @everywhere macro if they are to be visible by all worker processes. Every other MLJ construct should work fine.
pipe = MyPipe() #instantiate the model in the master process
#create a machine
mach = machine(pipe, X, height) #No need to use the @everywhere macro
#evaluate the machine
evaluate!(mach, resampling=Holdout(), measure=rms, verbosity=2, acceleration = MLJ.CPUProcesses())
#Note i used MLJ.CPUProcesses() instead of ComputationalResources.CPUProcesses() which is used in the docs just to tell one which package the type CPUProcesses() comes from. Similarly instead of MLJTuning.TunedModel use MLJ.TunedModel or just TunedModel.
Hope this helps.
PS:
Note that BLAS.set_num_threads applies multithreading to Matrix-vector operations.
OkonSamuel
on 26 Feb 2020
@OkonSamuel Thanks lot for this! I have used @everything in loading packages, including MLJ, but it seems couldn't find @load macro for some reason. What's the cause of this??
Code
@everywhere using MLJ
...
addprocs(32, exeflags=`--project=$@__DIR__`, enable_threaded_blas=true)
BLAS.set_num_threads(32)
...
@everywhere @load SubspaceLDA pkg=MultivariateStats
lda_model = SubspaceLDA()
...
Stacktrace
ERROR: LoadError: On worker 2:
LoadError: UndefVarError: @load not defined
top-level scope
eval at ./boot.jl:330
#105 at /buildworker/worker/package_linux64/build/usr/share/julia/stdlib/v1.3/Distributed/src/process_messages.jl:290
run_work_thunk at /buildworker/worker/package_linux64/build/usr/share/julia/stdlib/v1.3/Distributed/src/process_messages.jl:79
run_work_thunk at /buildworker/worker/package_linux64/build/usr/share/julia/stdlib/v1.3/Distributed/src/process_messages.jl:88
#98 at ./task.jl:333
in expression starting at /dmf/tri_services/Facilities/Imaging/Darren/DoD_classifier/compare_approaches.jl:191
...and 31 more exception(s).
Stacktrace:
[1] sync_end(::Array{Any,1}) at ./task.jl:300
[2] macro expansion at ./task.jl:319 [inlined]
[3] remotecall_eval(::Module, ::Array{Int64,1}, ::Expr) at /buildworker/worker/package_linux64/build/usr/share/julia/stdlib/v1.3/Distributed/src/macros.jl:217
[4] top-level scope at /buildworker/worker/package_linux64/build/usr/share/julia/stdlib/v1.3/Distributed/src/macros.jl:201
[5] include at ./boot.jl:328 [inlined]
[6] include_relative(::Module, ::String) at ./loading.jl:1105
[7] include(::Module, ::String) at ./Base.jl:31
[8] exec_options(::Base.JLOptions) at ./client.jl:287
[9] _start() at ./client.jl:460
in expression starting at /dmf/tri_services/Facilities/Imaging/Darren/DoD_classifier/compare_approaches.jl:191
@darrencl.
The LoadError you got is due to the order of execution of your code. You loaded MLJ on every available process (initially there was only 1 process) then you added more processes (which does not have MLJ loaded on them).
Instead you should have first added the number of needed processes (using Distributed; addprocs(32, exeflags=--project=$@__DIR__, enable_threaded_blas=true);
@everywhere using LinearAlgebra;
BLAS.set_num_threads(32)) before loading MLJ on all available process (@everywhere using MLJ).
If things were done this way there wouldn't be any error when executing @everywhere @load SubspaceLDA pkg=MultivariateStats.
OkonSamuel
on 29 Feb 2020
@OkonSamuel makes sense! Thanks a lot for the help! :)
darrencl
on 2 Mar 2020
Related issues
tlienart
·
6Comments
ChrisRackauckas
·
7Comments
ablaom
·
4Comments
CameronBieganek
·
6Comments
juliohm
·
7Comments
Most helpful comment
The high dimensionality of your problem is likely a cause of issues (p >>> n) i.e. LDA itself is likely super slow. Consider either doing logistic regression or subspace LDA, this quote is from multivariatestats:
https://github.com/JuliaStats/MultivariateStats.jl/blob/af1175c4c89043ea4a898dc4dcdbff68fe05d6f5/src/lda.jl#L203-L207
@OkonSamuel wrote an interface for it too so it should be a mere case of swapping the model name to see if that might bring a significant speedup