<jupyter_start><jupyter_text>Build and use function images (Kubernetes)This topic describes running a kubernetes-based job using shared data, and building custom container images.**In this section**- [Define a new function and its dependencies](define-a-new-function-and-its-dependencies)- [Convert the code to a serverless job](convert-the-Code-to-a-serverless-job)- [Deploy (build) the function container](deploy-build-the-function-container)- [Run the function on the cCluster](run-the-function-on-the-cluster)- [Create and run a Kubeflow pipeline](create-and-run-a-kubeflow-pipeline)  Define a new function and its dependenciesDefine a single serverless function with two `handlers`, one for training and one for validation.<jupyter_code>import mlrun 
import time
import pandas as pd
from mlrun.artifacts import get_model, update_model

def training(
    context,
    p1: int = 1,
    p2: int = 2
) -> None:
    """Train a model.

    :param context: The runtime context object.
    :param p1: A model parameter.
    :param p2: Another model parameter.
    """
    # access input metadata, values, and inputs
    print(f'Run: {context.name} (uid={context.uid})')
    print(f'Params: p1={p1}, p2={p2}')
    context.logger.info('started training')
    
    # <insert training code here>
    
    # log the run results (scalar values)
    context.log_result('accuracy', p1 * 2)
    context.log_result('loss', p1 * 3)
    
    # add a label/tag to this run 
    context.set_label('category', 'tests')
    
    # log a simple artifact + label the artifact 
    # If you want to upload a local file to the artifact repo add src_path=<local-path>
    context.log_artifact('somefile', 
                          body=b'abc is 123', 
                          local_path='myfile.txt')
    
    # create a dataframe artifact 
    df = pd.DataFrame([{'A':10, 'B':100}, {'A':11,'B':110}, {'A':12,'B':120}])
    context.log_dataset('mydf', df=df)
    
    # Log an ML Model artifact, add metrics, params, and labels to it
    # and place it in a subdir ('models') under artifacts path 
    context.log_model('mymodel', body=b'abc is 123', 
                      model_file='model.txt', 
                      metrics={'accuracy':0.85}, parameters={'xx':'abc'},
                      labels={'framework': 'xgboost'},
                      artifact_path=context.artifact_subpath('models'))

def validation(
    context,
    model: mlrun.DataItem
) -> None:
    """Model validation.
    
    Dummy validation function.
    
    :param context: The runtime context object.
    :param model: The extimated model object.
    """
    # access input metadata, values, files, and secrets (passwords)
    print(f'Run: {context.name} (uid={context.uid})')
    context.logger.info('started validation')
    
    # get the model file, class (metadata), and extra_data (dict of key: DataItem)
    model_file, model_obj, _ = get_model(model)

    # update model object elements and data
    update_model(model_obj, parameters={'one_more': 5})

    print(f'path to local copy of model file - {model_file}')
    print('parameters:', model_obj.parameters)
    print('metrics:', model_obj.metrics)
    context.log_artifact('validation', 
                         body=b'<b> validated </b>', 
                         format='html')<jupyter_output><empty_output><jupyter_text>The following end-code annotation tells MLRun to stop parsing the notebook from this cell. _**Do not remove this cell**_:<jupyter_code># mlrun: end-code<jupyter_output><empty_output><jupyter_text>______________________________________________  Convert the code to a serverless jobCreate a ```function``` that defines the runtime environment (type, code, image, ..) and ```run()``` a job or experiment using that function.In each run, you can specify the function, inputs, parameters/hyper-parameters, etc.Use the ```job``` runtime for running container jobs, or alternatively use another distributed runner like MpiJob, Spark, Dask, and Nuclio.**Setting up the environment**<jupyter_code>project_name, artifact_path = mlrun.set_environment(project='jobs-demo', artifact_path='./data/{{run.uid}}')<jupyter_output><empty_output><jupyter_text>**Define the cluster jobs, build images, and set dependencies**To use the function in a cluster you need to package the code and its dependencies.The ```code_to_function``` call automatically generates a ```function``` object from the current notebook (or specified file) with its list of dependencies and runtime configuration. In this example the code depends on the pandas package, so so it's specified in the  ```code_to_function``` call.<jupyter_code># create an ML function from the notebook, attach it to iguazio data fabric (v3io)
trainer = mlrun.code_to_function(name='my-trainer', kind='job', image='mlrun/mlrun', requirements=['pandas'])<jupyter_output><empty_output><jupyter_text>The functions need a shared storage media (file or object) to pass and store artifacts.You can add _**Kubernetes**_ resources like volumes, environment variables, secrets, cpu/mem/gpu, etc. to a function.```mlrun``` uses _**KubeFlow**_ modifiers (apply) to configure resources. You can build your own resources or use predefined resources e.g. [AWS resources](https://github.com/kubeflow/pipelines/blob/master/sdk/python/kfp/aws.py).The example above uses built-in images. When you move to production, use specific tags. For more details on built-in and custom images, see [MLRun images and external docker images](./images.htmlmlrun-images-and-external-docker-images).  **Option 1: Using file volumes for artifacts**MLRun automatically applies the most common storage configuration to functions. As a result, most cases do not require any additional storage configurations before executing a function. See more details in [Applying storage configurations to functions](./function-storage.md).If you're using the [Iguazio MLOps platform](https://www.iguazio.com/), and want to configure manually, use the `mount_v3io()` auto-mount modifier.If you're using another k8s PVC volume, use the `mlrun.platforms.mount_pvc(..)` modifier with the required parameters.This example uses the `auto_mount()` modifier. It auto-selects between the k8s PVC volume and the Iguazio data fabric. You can set the PVC volume configuration with the env var below or with the `auto_mount` params:```    MLRUN_PVC_MOUNT=:```If you apply `mount_v3io()` or `auto_mount()` when running the function in the MLOps platform, it attaches the function to Iguazio's real-time data fabric (mounted by default to _**home**_ of the current user).**Note**: If the notebook is not on the managed platform (it's running remotely) you might need to use secrets. For the current ```training``` function, run:<jupyter_code># for PVC volumes set the env var for PVC: MLRUN_PVC_MOUNT=<pvc-name>:<mount-path>, pass the relevant parameters
from mlrun.platforms import auto_mount
trainer.apply(auto_mount())<jupyter_output><empty_output><jupyter_text>_**Option 2: Using AWS S3 for artifacts**_ When using AWS, you can use S3. See more details in [S3](../store/datastore.htmls3).  Deploy (build) the function containerThe `deploy()` command builds a custom container image (creates a cluster build job) from the outlined function dependencies.If a pre-built container image already exists, pass the `image` name instead. _**The code and params can be updated per run without building a new image**_.The image is stored in a container repository. By default it uses the repository configured on the MLRun API service. You can specify your own docker registry by first creating a secret, and adding that secret name to the build configuration: `kubectl create -n  secret docker-registry my-docker --docker-server=https://index.docker.io/v1/ --docker-username= --docker-password= --docker-email=` And then run this: `trainer.build_config(image='target/image:tag', secret='my_docker')`<jupyter_code>trainer.deploy(with_mlrun=False)<jupyter_output>> 2021-01-24 00:05:18,384 [info] starting remote build, image: .mlrun/func-jobs-demo-my-trainer-latest
[36mINFO[0m[0020] Retrieving image manifest mlrun/mlrun:unstable 
[36mINFO[0m[0020] Retrieving image manifest mlrun/mlrun:unstable 
[36mINFO[0m[0021] Built cross stage deps: map[]                
[36mINFO[0m[0021] Retrieving image manifest mlrun/mlrun:unstable 
[36mINFO[0m[0021] Retrieving image manifest mlrun/mlrun:unstable 
[36mINFO[0m[0021] Executing 0 build triggers                   
[36mINFO[0m[0021] Unpacking rootfs as cmd RUN pip install pandas requires it. 
[36mINFO[0m[0037] RUN pip install pandas                       
[36mINFO[0m[0037] Taking snapshot of full filesystem...        
[36mINFO[0m[0050] cmd: /bin/sh                                 
[36mINFO[0m[0050] args: [-c pip install pandas]                
[36mINFO[0m[0050] Running: [/bin/sh -c pip install pandas]     
Requirement already satisfied: pandas in /usr/local/lib/python3.7/site-packages (1[...]<jupyter_text>Run the function on the clusterUse ```with_code``` to inject the latest code into the function (without requiring a new build).<jupyter_code>trainer.with_code()
# run our training task with params
train_run = trainer.run(name='my-training', handler='training', params={'p1': 9})
# running validation, use the model result from the previous step 
model = train_run.outputs['mymodel']
validation_run = trainer.run(name='validation', handler='validation', inputs={'model': model}, watch=True)<jupyter_output>> 2021-01-24 00:09:21,259 [info] starting run validation uid=c757ffcdc36d4412b4bcba1df75f079d DB=http://mlrun-api:8080
> 2021-01-24 00:09:21,536 [info] Job is running in the background, pod: validation-dwd78
> 2021-01-24 00:09:25,570 [warning] Unable to parse server or client version. Assuming compatible: {'server_version': 'unstable', 'client_version': 'unstable'}
Run: validation (uid=c757ffcdc36d4412b4bcba1df75f079d)
> 2021-01-24 00:09:25,719 [info] started validation
path to local copy of model file - /User/data/30b8131285a74f87b16d957fabc5fac3/models/model.txt
parameters: {'xx': 'abc', 'one_more': 5}
metrics: {'accuracy': 0.85}
> 2021-01-24 00:09:25,873 [info] run executed, status=completed
final state: completed<jupyter_text>Create and run a Kubeflow pipelineKubeflow pipelines are used for workflow automation, creating a graph of functions and their specified parameters, inputs, and outputs.You can chain the outputs and inputs of the pipeline steps, as illustrated below.<jupyter_code>import kfp
from kfp import dsl
from mlrun import run_pipeline
from mlrun import run_function, deploy_function
@dsl.pipeline(
    name = 'job test',
    description = 'demonstrating mlrun usage'
)
def job_pipeline(
   p1: int = 9
) -> None:
    """Define our pipeline.
    
    :param p1: A model parameter.
    """

    train = run_function('my-trainer',
                            handler='training',
                            params={'p1': p1},
                            outputs=['mymodel'])
    
    validate = run_function('my-trainer',
                               handler='validation',
                               inputs={'model': train.outputs['mymodel']},
                               outputs=['validation'])<jupyter_output><empty_output><jupyter_text>Running the pipeline Pipeline results are stored at the `artifact_path` location. The artifact path for workflows can be one of:- The project's `artifact_path` (set by `project.spec.artifact_path = ''`). MLRun adds `/{{workflow.uid}}` to the path if it does not already include it.- MLRun's default `artifact-path`, if set.  MLRun adds `/{{workflow.uid}}`' to the path if it does not already include it.- The `artifact_path` as passed to the specific call for `run()`, as shown below. In this case, MLRun does not modify the user-provided path.  If you want to customize the path, per workflow, use:<jupyter_code>artifact_path = 'v3io:///users/admin/kfp/{{workflow.uid}}/'
arguments = {'p1': 8}
run_id = run_pipeline(job_pipeline, arguments, experiment='my-job', artifact_path=artifact_path)
from mlrun import wait_for_pipeline_completion, get_run_db
wait_for_pipeline_completion(run_id)
db = get_run_db().list_runs(project=project_name, labels=f'workflow={run_id}').show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11004317015557546, 'fraction_numerical': 0.025494827726643316, 'mean_word_length': 4.189349112426036, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 50, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '284129', 'license': 'Apache-2.0', 'n_tokens_mistral': 3852, 'n_tokens_neox': 3523, 'n_words': 1230, 'path': 'docs/runtimes/mlrun_jobs.ipynb', 'repo_name': 'george0st/mlrun', 'size': '12277'}