lynnmatrix / kfp-local Goto Github PK

ERROR:root:Traceback (most recent call last):
  File "opt/anaconda3/envs/test/lib/python3.8/runpy.py", line 194, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "opt/anaconda3/envs/test/lib/python3.8/runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "opt/anaconda3/envs/test/lib/python3.8/site-packages/kfp/v2/components/executor_main.py", line 104, in <module>
    executor_main()
  File "opt/anaconda3/envs/test/lib/python3.8/site-packages/kfp/v2/components/executor_main.py", line 97, in executor_main
    executor = component_executor.Executor(
  File "opt/anaconda3/envs/test/lib/python3.8/site-packages/kfp/v2/components/executor.py", line 46, in __init__
    self._output_artifacts[name] = self._make_output_artifact(
  File "opt/anaconda3/envs/test/lib/python3.8/site-packages/kfp/v2/components/executor.py", line 61, in _make_output_artifact
    os.makedirs(os.path.dirname(artifact.path), exist_ok=True)
  File "opt/anaconda3/envs/test/lib/python3.8/os.py", line 213, in makedirs
    makedirs(head, exist_ok=exist_ok)
  File "opt/anaconda3/envs/test/lib/python3.8/os.py", line 213, in makedirs
    makedirs(head, exist_ok=exist_ok)
  File "opt/anaconda3/envs/test/lib/python3.8/os.py", line 213, in makedirs
    makedirs(head, exist_ok=exist_ok)
  [Previous line repeated 4 more times]
  File "opt/anaconda3/envs/test/lib/python3.8/os.py", line 223, in makedirs
    mkdir(name, mode)
OSError: [Errno 30] Read-only file system: '/s3'

ERROR:root:['sh', '-c', '\n\nif ! [ -x "$(command -v pip)" ]; then\n    python3 -m ensurepip || python3 -m ensurepip --user || apt-get install python3-pip\nfi\n\nPIP_DISABLE_PIP_VERSION_CHECK=1 python3 -m pip install --quiet     --no-warn-script-location \'pandas\' \'kfp==1.8.13\' && "$0" "$@"\n', 'sh', '-ec', 'program_path=$(mktemp -d)\nprintf "%s" "$0" > "$program_path/ephemeral_component.py"\npython3 -m kfp.v2.components.executor_main                         --component_module_path                         "$program_path/ephemeral_component.py"                         "$@"\n', '\nimport kfp\nfrom kfp.v2 import dsl\nfrom kfp.v2.dsl import *\nfrom typing import *\n\ndef build_input_dataset(\n        output_data: Output[Dataset]):\n    """Generates iris dataset for training"""\n    import pandas as pd\n    data = pd.DataFrame({\n         "calories": [420, 380, 390],\n         "duration": [50, 40, 45]\n    })\n\n    # Write the dataset to GCS location as Output that will be Input to the next step\n    data.to_csv(output_data.path)\n\n', '--executor_input', '{\n  "inputs": {},\n  "outputs": {\n    "artifacts": {\n      "output_data": {\n        "artifacts": [\n          {\n            "type": {\n              "schemaTitle": "system.Dataset",\n              "schemaVersion": "0.0.1"\n            },\n            "uri": "s3://var/folders/m9/pt1xjxh10c7__vy9tf_gygw00000gn/T/default-pipeline-template_20230103222819/build-input-dataset/output_data",\n            "metadata": {}\n          }\n        ]\n      }\n    },\n    "outputFile": "/var/folders/m9/pt1xjxh10c7__vy9tf_gygw00000gn/T/default-pipeline-template_20230103222819/build-input-dataset/output_metadata.json"\n  }\n}', '--function_to_execute', 'build_input_dataset']

from kfp.v2 import dsl
from kfp.v2.dsl import Dataset, Output, component
from kfp_local import LocalClient

@component(
    packages_to_install=["pandas"],
    base_image="python:3.8",
)
def build_input_dataset(
        output_data: Output[Dataset]):
    """Generates iris dataset for training"""
    import pandas as pd
    data = pd.DataFrame({
         "calories": [420, 380, 390],
         "duration": [50, 40, 45]
    })

    # Write the dataset to GCS location as Output that will be Input to the next step
    data.to_csv(output_data.path)


@dsl.pipeline(name='default-pipeline-template')
def pipeline():
    """DAG that defines the kubeflow pipeline workflow"""
    dataset_create_op = build_input_dataset()

local_client = LocalClient()

result = local_client.create_run_from_pipeline_func(
    pipeline,
    arguments={},
    execution_mode=LocalClient.ExecutionMode("local"),
)
if result.success:
    print("Local run successfully")