Merge pull request #16 from JuliaAI/listartifacts

listartifacts

Author: deyandyankov

Project.toml

@@ -13,10 +13,10 @@ URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4"
 UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
 
 [compat]
-ShowCases = "0.1"
 FilePathsBase = "0.9"
 HTTP = "0.9"
 JSON = "0.21"
+ShowCases = "0.1"
 URIs = "1"
 julia = "1"
 

docs/src/reference.md

@@ -16,6 +16,8 @@ MLFlowRunInfo
 MLFlowRunData
 MLFlowRunDataMetric
 MLFlowRunStatus
+MLFlowArtifactFileInfo
+MLFlowArtifactDirInfo
 ```
 
 # Experiments
@@ -36,14 +38,10 @@ getrun
 updaterun
 deleterun
 searchruns
-```
-
-# Logging
-
-```@docs
 logparam
 logmetric
 logartifact
+listartifacts
 ```
 
 # Utilities

src/MLFlowClient.jl

@@ -32,7 +32,11 @@ export
     MLFlowRunDataMetric,
     MLFlowRun,
     get_info,
-    get_data
+    get_data,
+    MLFlowArtifactFileInfo,
+    MLFlowArtifactDirInfo,
+    get_path,
+    get_size
 
 include("utils.jl")
 export
@@ -52,12 +56,10 @@ export
     getrun,
     updaterun,
     deleterun,
-    searchruns
-
-include("logging.jl")
-export
+    searchruns,
     logparam,
     logmetric,
-    logartifact
+    logartifact,
+    listartifacts
 
 end

src/logging.jl

@@ -184,3 +184,196 @@ searchruns(mlf::MLFlow, exp::MLFlowExperiment; kwargs...) =
     searchruns(mlf, exp.experiment_id; kwargs...)
 searchruns(mlf::MLFlow, exps::AbstractVector{MLFlowExperiment}; kwargs...) =
     searchruns(mlf, [getfield.(exps, :experiment_id)]; kwargs...)
+
+
+"""
+    logparam(mlf::MLFlow, run, key, value)
+    logparam(mlf::MLFlow, run, kv)
+
+Associates a key/value pair of parameters to the particular run.
+
+# Arguments
+- `mlf`: [`MLFlow`](@ref) configuration.
+- `run`: one of [`MLFlowRun`](@ref), [`MLFlowRunInfo`](@ref), or `String`.
+- `key`: parameter key (name). Automatically converted to string before sending to MLFlow because this is the only type that MLFlow supports.
+- `value`: parameter value. Automatically converted to string before sending to MLFlow because this is the only type that MLFlow supports.
+
+One could also specify `kv::Dict` instead of separate `key` and `value` arguments.
+"""
+function logparam(mlf::MLFlow, run_id::String, key, value)
+    endpoint = "runs/log-parameter"
+    mlfpost(mlf, endpoint; run_id=run_id, key=string(key), value=string(value))
+end
+logparam(mlf::MLFlow, run_info::MLFlowRunInfo, key, value) =
+    logparam(mlf, run_info.run_id, key, value)
+logparam(mlf::MLFlow, run::MLFlowRun, key, value) =
+    logparam(mlf, run.info, key, value)
+function logparam(mlf::MLFlow, run::Union{String,MLFlowRun,MLFlowRunInfo}, kv)
+    for (k, v) in kv
+        logparam(mlf, run, k, v)
+    end
+end
+
+"""
+    logmetric(mlf::MLFlow, run, key, value::T; timestamp, step) where T<:Real
+    logmetric(mlf::MLFlow, run, key, values::AbstractArray{T}; timestamp, step) where T<:Real
+
+Logs a metric value (or values) against a particular run.
+
+# Arguments
+- `mlf`: [`MLFlow`](@ref) configuration.
+- `run`: one of [`MLFlowRun`](@ref), [`MLFlowRunInfo`](@ref), or `String`
+- `key`: metric name.
+- `value`: metric value, must be numeric.
+
+# Keywords
+- `timestamp`: if provided, must be a UNIX timestamp in milliseconds. By default, set to current time.
+- `step`: step at which the metric value has been taken.
+"""
+function logmetric(mlf::MLFlow, run_id::String, key, value::T; timestamp=missing, step=missing) where T<:Real
+    endpoint = "runs/log-metric"
+    if ismissing(timestamp)
+        timestamp = Int(trunc(datetime2unix(now()) * 1000))
+    end
+    mlfpost(mlf, endpoint; run_id=run_id, key=key, value=value, timestamp=timestamp, step=step)
+end
+logmetric(mlf::MLFlow, run_info::MLFlowRunInfo, key, value::T; timestamp=missing, step=missing) where T<:Real =
+    logmetric(mlf::MLFlow, run_info.run_id, key, value; timestamp=timestamp, step=step)
+logmetric(mlf::MLFlow, run::MLFlowRun, key, value::T; timestamp=missing, step=missing) where T<:Real =
+    logmetric(mlf, run.info, key, value; timestamp=timestamp, step=step)
+
+function logmetric(mlf::MLFlow, run::Union{String,MLFlowRun,MLFlowRunInfo}, key, values::AbstractArray{T}; timestamp=missing, step=missing) where T<:Real
+    for v in values
+        logmetric(mlf, run, key, v; timestamp=timestamp, step=step)
+    end
+end
+
+
+"""
+    logartifact(mlf::MLFlow, run, basefilename, data)
+
+Stores an artifact (file) in the run's artifact location.
+
+!!! note
+    Assumes that artifact_uri is mapped to a local directory.
+    At the moment, this only works if both MLFlow and the client are running on the same host or they map a directory that leads to the same location over NFS, for example.
+
+# Arguments
+- `mlf::MLFlow`: [`MLFlow`](@ref) onfiguration. Currently not used, but when this method is extended to support `S3`, information from `mlf` will be needed.
+- `run`: one of [`MLFlowRun`](@ref), [`MLFlowRunInfo`](@ref) or `String`.
+- `basefilename`: name of the file to be written.
+- `data`: artifact content, an object that can be written directly to a file handle.
+
+# Throws
+- an `ErrorException` if an exception occurs during writing artifact.
+
+# Returns
+path of the artifact that was created.
+"""
+function logartifact(mlf::MLFlow, run_id::AbstractString, basefilename::AbstractString, data)
+    mlflowrun = getrun(mlf, run_id)
+    artifact_uri = mlflowrun.info.artifact_uri
+    mkpath(artifact_uri)
+    filepath = joinpath(artifact_uri, basefilename)
+    try
+        f = open(filepath, "w")
+        write(f, data)
+        close(f)
+    catch e
+        error("Unable to create artifact $(filepath): $e")
+    end
+    filepath
+end
+logartifact(mlf::MLFlow, run::MLFlowRun, basefilename::AbstractString, data) =
+    logartifact(mlf, run.info, basefilename, data)
+logartifact(mlf::MLFlow, run_info::MLFlowRunInfo, basefilename::AbstractString, data) =
+    logartifact(mlf, run_info.run_id, basefilename, data)
+
+"""
+    logartifact(mlf::MLFlow, run, filepath)
+
+Stores an artifact (file) in the run's artifact location.
+The name of the artifact is calculated using `basename(filepath)`.
+
+Dispatches on `logartifact(mlf::MLFlow, run, basefilename, data)` where `data` is the contents of `filepath`.
+
+# Throws
+- an `ErrorException` if `filepath` does not exist.
+- an exception if such occurs while trying to read the contents of `filepath`.
+
+"""
+function logartifact(mlf::MLFlow, run_id::AbstractString, filepath::Union{AbstractPath,AbstractString})
+    isfile(filepath) || error("File $filepath does not exist.")
+    try
+        f = open(filepath, "r")
+        data = read(f)
+        close(f)
+        return logartifact(mlf, run_id, basename(filepath), data)
+    catch e
+        throw(e)
+    finally
+        if @isdefined f
+            close(f)
+        end
+    end
+end
+logartifact(mlf::MLFlow, run::MLFlowRun, filepath::Union{AbstractPath,AbstractString}) =
+    logartifact(mlf, run.info, filepath)
+logartifact(mlf::MLFlow, run_info::MLFlowRunInfo, filepath::Union{AbstractPath,AbstractString}) =
+    logartifact(mlf, run_info.run_id, filepath)
+
+"""
+    listartifacts(mlf::MLFlow, run)
+
+Lists the artifacts associated with an experiment run.
+According to [MLFlow documentation](https://mlflow.org/docs/latest/rest-api.html#list-artifacts), this API endpoint should return paged results, similar to [`searchruns`](@ref).
+However, after some experimentation, this doesn't seem to be the case. Therefore, the paging functionality is not implemented here.
+
+# Arguments
+- `mlf::MLFlow`: [`MLFlow`](@ref) onfiguration. Currently not used, but when this method is extended to support `S3`, information from `mlf` will be needed.
+- `run`: one of [`MLFlowRun`](@ref), [`MLFlowRunInfo`](@ref) or `String`.
+
+# Keywords
+- `path::String`: path of a directory within the artifact location. If set, returns the contents of the directory. By default, this is the root directory of the artifacts.
+- `maxdepth::Int64`: depth of listing. Default is 1. This will only return the files/directories in the current `path`. To return all artifacts files and directories, use `maxdepth=-1`.
+
+# Returns
+A vector of `Union{MLFlowArtifactFileInfo,MLFlowArtifactDirInfo}`.
+"""
+function listartifacts(mlf::MLFlow, run_id::String; path::String="", maxdepth::Int64=1)
+    endpoint = "artifacts/list"
+    kwargs = (
+        run_id=run_id,
+    )
+    kwargs = (; kwargs..., path=path)
+    httpresult = mlfget(mlf, endpoint; kwargs...)
+    "files" ∈ keys(httpresult) || return Vector{Union{MLFlowArtifactFileInfo,MLFlowArtifactDirInfo}}()
+    "root_uri" ∈ keys(httpresult) || error("Malformed response from MLFlow REST API.")
+    root_uri = httpresult["root_uri"]
+    result = Vector{Union{MLFlowArtifactFileInfo,MLFlowArtifactDirInfo}}()
+    maxdepth == 0 && return result
+
+    for resultentry ∈ httpresult["files"]
+        if resultentry["is_dir"] == false
+            filepath = joinpath(root_uri, resultentry["path"])
+            filesize = parse(Int, resultentry["file_size"])
+            push!(result, MLFlowArtifactFileInfo(filepath, filesize))
+        elseif resultentry["is_dir"] == true
+            dirpath = joinpath(root_uri, resultentry["path"])
+            push!(result, MLFlowArtifactDirInfo(dirpath))
+            if maxdepth != 0
+                nextdepthresult = listartifacts(mlf, run_id, path=resultentry["path"], maxdepth=maxdepth-1)
+                result = vcat(result, nextdepthresult)
+            end
+        else
+            isdirval = resultentry["is_dir"]
+            @warn "Malformed response from MLFlow REST API is_dir=$isdirval - skipping"
+            continue
+        end
+    end
+    result
+end
+listartifacts(mlf::MLFlow, run::MLFlowRun; kwargs...) =
+    listartifacts(mlf, run.info.run_id; kwargs...)
+listartifacts(mlf::MLFlow, run_info::MLFlowRunInfo; kwargs...) =
+    listartifacts(mlf, run_info.run_id; kwargs...)