Some fixes on experiments. Adding the run service and misc additions

Author: pebeto

src/MLFlowClient.jl

@@ -60,4 +60,9 @@ export
     restoreexperiment,
     searchexperiments,
     getexperimentbyname
+
+include("services/run.jl")
+export 
+    createrun
+
 end

src/deprecated.jl

@@ -1,5 +1,5 @@
 """
-    createexperiment(instance::MLFlow; name::String="",
+    createexperiment(instance::MLFlow, name::String;
         artifact_location::String="",
         tags::Union{Dict{<:Any}, Array{<:Any}}=[])
 
@@ -18,13 +18,9 @@ default.
 # Returns
 The ID of the newly created experiment.
 """
-function createexperiment(instance::MLFlow; name::String="",
-    artifact_location::String="",
+function createexperiment(instance::MLFlow, name::String;
+    artifact_location::Union{String, Missing}=missing,
     tags::Union{Dict{<:Any}, Array{<:Any}}=[])::String
-    if name |> isempty
-        name = UUIDs.uuid4() |> string
-    end
-
     tags = tags |> parsetags
 
     try
@@ -100,26 +96,26 @@ function getexperimentbyname(instance::MLFlow, experiment_name::String)
 end
 
 """
-    deleteexperiment(mlf::MLFlow, experiment_id::String)
-    deleteexperiment(mlf::MLFlow, experiment_id::Integer)
-    deleteexperiment(mlf::MLFlow, experiment::Experiment)
+    deleteexperiment(instance::MLFlow, experiment_id::String)
+    deleteexperiment(instance::MLFlow, experiment_id::Integer)
+    deleteexperiment(instance::MLFlow, experiment::Experiment)
 
 Mark an experiment and associated metadata, runs, metrics, params, and tags for
 deletion. If the experiment uses FileStore, artifacts associated with
 experiment are also deleted.
 
 # Arguments
-- `mlf`: [`MLFlow`](@ref) configuration.
+- `instance`: [`MLFlow`](@ref) configuration.
 - `experiment_id`: ID of the associated experiment.
 
 # Returns
 
 `true` if successful. Otherwise, raises exception.
 """
-function deleteexperiment(mlf::MLFlow, experiment_id::String)
+function deleteexperiment(instance::MLFlow, experiment_id::String)
     endpoint = "experiments/delete"
     try
-        mlfpost(mlf, endpoint; experiment_id=experiment_id)
+        mlfpost(instance, endpoint; experiment_id=experiment_id)
         return true
     catch e
         if isa(e, HTTP.ExceptionRequest.StatusError) && e.status == 404
@@ -129,32 +125,32 @@ function deleteexperiment(mlf::MLFlow, experiment_id::String)
         throw(e)
     end
 end
-deleteexperiment(mlf::MLFlow, experiment_id::Integer) =
-    deleteexperiment(mlf, string(experiment_id))
-deleteexperiment(mlf::MLFlow, experiment::Experiment) =
-    deleteexperiment(mlf, experiment.experiment_id)
+deleteexperiment(instance::MLFlow, experiment_id::Integer) =
+    deleteexperiment(instance, string(experiment_id))
+deleteexperiment(instance::MLFlow, experiment::Experiment) =
+    deleteexperiment(instance, experiment.experiment_id)
 
 """
-    restoreexperiment(mlf::MLFlow, experiment_id::String)
-    restoreexperiment(mlf::MLFlow, experiment_id::Integer)
-    restoreexperiment(mlf::MLFlow, experiment::Experiment)
+    restoreexperiment(instance::MLFlow, experiment_id::String)
+    restoreexperiment(instance::MLFlow, experiment_id::Integer)
+    restoreexperiment(instance::MLFlow, experiment::Experiment)
 
 Restore an experiment marked for deletion. This also restores associated
 metadata, runs, metrics, params, and tags. If experiment uses FileStore,
 underlying artifacts associated with experiment are also restored.
 
 # Arguments
-- `mlf`: [`MLFlow`](@ref) configuration.
+- `instance`: [`MLFlow`](@ref) configuration.
 - `experiment_id`: ID of the associated experiment.
 
 # Returns
 
 `true` if successful. Otherwise, raises exception.
 """
-function restoreexperiment(mlf::MLFlow, experiment_id::String)
+function restoreexperiment(instance::MLFlow, experiment_id::String)
     endpoint = "experiments/restore"
     try
-        mlfpost(mlf, endpoint; experiment_id=experiment_id)
+        mlfpost(instance, endpoint; experiment_id=experiment_id)
         return true
     catch e
         if isa(e, HTTP.ExceptionRequest.StatusError) && e.status == 404
@@ -166,48 +162,51 @@ function restoreexperiment(mlf::MLFlow, experiment_id::String)
         throw(e)
     end
 end
-restoreexperiment(mlf::MLFlow, experiment_id::Integer) =
-    deleteexperiment(mlf, string(experiment_id))
-restoreexperiment(mlf::MLFlow, experiment::Experiment) =
-    deleteexperiment(mlf, experiment.experiment_id)
+restoreexperiment(instance::MLFlow, experiment_id::Integer) =
+    deleteexperiment(instance, string(experiment_id))
+restoreexperiment(instance::MLFlow, experiment::Experiment) =
+    deleteexperiment(instance, experiment.experiment_id)
 
 """
-    updateexperiment(mlf::MLFlow, experiment_id::String, new_name::String)
-    updateexperiment(mlf::MLFlow, experiment_id::Integer, new_name::String)
-    updateexperiment(mlf::MLFlow, experiment::Experiment, new_name::String)
+    updateexperiment(instance::MLFlow, experiment_id::String, new_name::String)
+    updateexperiment(instance::MLFlow, experiment_id::Integer,
+        new_name::String)
+    updateexperiment(instance::MLFlow, experiment::Experiment,
+        new_name::String)
 
 Update experiment metadata.
 
 # Arguments
-- `mlf`: [`MLFlow`](@ref) configuration.
+- `instance`: [`MLFlow`](@ref) configuration.
 - `experiment_id`: ID of the associated experiment.
 - `new_name`: If provided, the experiment’s name is changed to the new name.
 The new name must be unique.
 
 # Returns
 `true` if successful. Otherwise, raises exception.
 """
-function updateexperiment(mlf::MLFlow, experiment_id::String, new_name::String)
+function updateexperiment(instance::MLFlow, experiment_id::String,
+    new_name::String)
     endpoint = "experiments/update"
     try
-        mlfpost(mlf, endpoint; experiment_id=experiment_id, new_name=new_name)
+        mlfpost(instance, endpoint; experiment_id=experiment_id, new_name=new_name)
         return true
     catch e
         throw(e)
     end
 end
-updateexperiment(mlf::MLFlow, experiment_id::Integer, new_name::String) =
-    updateexperiment(mlf, string(experiment_id), new_name)
-updateexperiment(mlf::MLFlow, experiment::Experiment, new_name::String) =
-    updateexperiment(mlf, experiment.experiment_id, new_name::String)
+updateexperiment(instance::MLFlow, experiment_id::Integer, new_name::String) =
+    updateexperiment(instance, string(experiment_id), new_name)
+updateexperiment(instance::MLFlow, experiment::Experiment, new_name::String) =
+    updateexperiment(instance, experiment.experiment_id, new_name::String)
 
 """
-    searchexperiments(mlf::MLFlow; max_results::Integer=20000,
+    searchexperiments(instance::MLFlow; max_results::Integer=20000,
         page_token::String="", filter::String="", order_by::Array{String}=[],
         view_type::ViewType=ACTIVE_ONLY)
 
 # Arguments
-- `mlf`: [`MLFlow`](@ref) configuration.
+- `instance`: [`MLFlow`](@ref) configuration.
 - `max_results`: Maximum number of experiments desired.
 - `page_token`: Token indicating the page of experiments to fetch.
 - `filter`: A filter expression over experiment attributes and tags that allows
@@ -221,9 +220,9 @@ unspecified, return only active experiments.
 # Returns
 - vector of [`MLFlowExperiment`](@ref) experiments that were found in the MLFlow instance
 """
-function searchexperiments(mlf::MLFlow; max_results::Integer=20000,
+function searchexperiments(instance::MLFlow; max_results::Integer=20000,
     page_token::String="", filter::String="", order_by::Array{String}=String[],
-    view_type::ViewType=ACTIVE_ONLY)
+    view_type::ViewType=ACTIVE_ONLY)::Tuple{Array{Experiment}, Union{String, Nothing}}
     endpoint = "experiments/search"
     parameters = (; max_results, page_token, filter,
         :view_type => view_type |> Integer)
@@ -233,8 +232,12 @@ function searchexperiments(mlf::MLFlow; max_results::Integer=20000,
     end
 
     try
-        result = mlfget(mlf, endpoint; parameters...)
-        return result["experiments"] |> (x -> [Experiment(y) for y in x])
+        result = mlfget(instance, endpoint; parameters...)
+
+        experiments = result["experiments"] |> (x -> [Experiment(y) for y in x])
+        next_page_token = get(result, "next_page_token", nothing)
+
+        return experiments, next_page_token
     catch e
         throw(e)
     end

src/services/experiment.jl

@@ -0,0 +1,33 @@
+"""
+    createrun(instance::MLFlow, experiment_id::String;
+        run_name::Union{String, Missing}=missing,
+        start_time::Union{Integer, Missing}=missing,
+        tags::Union{Dict{<:Any}, Array{<:Any}}=[])
+
+Create a new run within an experiment. A run is usually a single execution of a
+machine learning or data ETL pipeline.
+
+# Arguments
+- `instance`: [`MLFlow`](@ref) configuration.
+- `experiment_id`: ID of the associated experiment.
+- `run_name`: Name of the run.
+- `start_time`: Unix timestamp in milliseconds of when the run started.
+- `tags`: Additional metadata for run.
+
+# Returns
+An instance of type [`Run`](@ref).
+"""
+function createrun(instance::MLFlow, experiment_id::String;
+    run_name::Union{String, Missing}=missing,
+    start_time::Union{Integer, Missing}=missing,
+    tags::Union{Dict{<:Any}, Array{<:Any}}=[])
+    tags = tags |> parsetags
+
+    try
+        result = mlfpost(instance, "runs/create"; experiment_id=experiment_id,
+            run_name=run_name, start_time=start_time, tags=tags)
+        return result["run"] |> Run
+    catch e
+        throw(e)
+    end
+end

src/services/run.jl

@@ -24,8 +24,9 @@ struct Dataset
     schema::Union{String, Nothing}
     profile::Union{String, Nothing}
 end
-Dataset(name, digest, source_type, source; schema=nothing, profile=nothing) =
-    Dataset(name, digest, source_type, source, schema, profile)
+Dataset(data::Dict{String, Any}) = Dataset(
+    data["name"], data["digest"], data["source_type"], data["source"],
+    get(data, "schema", nothing), get(data, "profile", nothing))
 Base.show(io::IO, t::Dataset) = show(io, ShowCase(t, new_lines=true))
 
 """
@@ -45,5 +46,7 @@ struct DatasetInput
     tags::Array{Tag}
     dataset::Dataset
 end
-DatasetInput(dataset; tags=[]) = DatasetInput(tags, dataset)
+DatasetInput(data::Dict{String, Any}) = DatasetInput(
+    [Tag(tag) for tag in get(data, "tags", [])],
+    Dataset(data["dataset"]))
 Base.show(io::IO, t::DatasetInput) = show(io, ShowCase(t, new_lines=true))

src/types/dataset.jl

@@ -32,6 +32,7 @@ Status of a run.
     FAILED=4
     KILLED=5
 end
+RunStatus(status::String) = Dict(value => key for (key, value) in RunStatus |> Base.Enums.namemap)[status |> Symbol] |> RunStatus
 
 """
     ViewType

src/types/enums.jl

@@ -57,10 +57,14 @@ struct RunInfo
     experiment_id::String
     status::RunStatus
     start_time::Int64
-    end_time::Int64
+    end_time::Union{Int64, Nothing}
     artifact_uri::String
     lifecycle_stage::String
 end
+RunInfo(data::Dict{String, Any}) = RunInfo(data["run_id"], data["run_name"],
+    data["experiment_id"], RunStatus(data["status"]), data["start_time"],
+    get(data, "end_time", nothing), data["artifact_uri"],
+    data["lifecycle_stage"])
 Base.show(io::IO, t::RunInfo) = show(io, ShowCase(t, new_lines=true))
 
 """
@@ -78,6 +82,10 @@ struct RunData
     params::Array{Param}
     tags::Array{Tag}
 end
+RunData(data::Dict{String, Any}) = RunData(
+    [Metric(metric) for metric in get(data, "metrics", [])],
+    [Param(param) for param in get(data, "params", [])],
+    [Tag(tag) for tag in get(data, "tags", [])])
 Base.show(io::IO, t::RunData) = show(io, ShowCase(t, new_lines=true))
 
 """
@@ -91,6 +99,9 @@ Run inputs.
 struct RunInputs
     dataset_inputs::Array{DatasetInput}
 end
+RunInputs(data::Dict{String, Any}) = RunInputs(
+    [DatasetInput(dataset_input) for dataset_input in
+        get(data, "dataset_inputs", [])])
 Base.show(io::IO, t::RunInputs) = show(io, ShowCase(t, new_lines=true))
 
 """
@@ -103,4 +114,6 @@ struct Run
     data::RunData
     inputs::RunInputs
 end
+Run(data::Dict{String, Any}) = Run(RunInfo(data["info"]),
+    RunData(data["data"]), RunInputs(data["inputs"]))
 Base.show(io::IO, t::Run) = show(io, ShowCase(t, new_lines=true))

src/types/run.jl

@@ -1,43 +1,5 @@
 IntOrString = Union{Int, String}
 
-"""
-    generatefilterfromentity_type(filter_params::AbstractDict{K,V}, entity_type::String) where {K,V}
-
-Generates a `filter` string from `filter_params` dictionary and `entity_type`.
-
-# Arguments
-- `filter_params`: dictionary to use for filter generation.
-- `entity_type`: entity type to use for filter generation.
-
-# Returns
-A string that can be passed as `filter` to [`searchruns`](@ref).
-
-# Examples
-
-```@example
-generatefilterfromentity_type(Dict("paramkey1" => "paramvalue1", "paramkey2" => "paramvalue2"), "param")
-```
-"""
-function generatefilterfromentity_type(filter_params::AbstractDict{K,V}, entity_type::String) where {K,V}
-    length(filter_params) > 0 || return ""
-    # NOTE: may have issues with escaping.
-    filters = ["$(entity_type).\"$(k)\" = \"$(v)\"" for (k, v) ∈ filter_params]
-    join(filters, " and ")
-end
-
-"""
-    generatefilterfromparams(filter_params::AbstractDict{K,V}) where {K,V}
-
-Generates a `filter` string from `filter_params` dictionary and `param` entity type.
-"""
-generatefilterfromparams(filter_params::AbstractDict{K,V}) where {K,V} = generatefilterfromentity_type(filter_params, "param")
-"""
-    generatefilterfrommattributes(filter_attributes::AbstractDict{K,V}) where {K,V}
-
-Generates a `filter` string from `filter_attributes` dictionary and `attribute` entity type.
-"""
-generatefilterfromattributes(filter_attributes::AbstractDict{K,V}) where {K,V} = generatefilterfromentity_type(filter_attributes, "attribute")
-
 const MLFLOW_ERROR_CODES = (;
     RESOURCE_ALREADY_EXISTS = "RESOURCE_ALREADY_EXISTS",
     RESOURCE_DOES_NOT_EXIST = "RESOURCE_DOES_NOT_EXIST",

src/utils.jl

@@ -4,8 +4,8 @@ end
 
 include("base.jl")
 
-include("services/experiments.jl")
+include("services/experiment.jl")
+include("services/run.jl")
 # include("test_functional.jl")
-# include("test_experiments.jl")
 # include("test_runs.jl")
 # include("test_loggers.jl")