<jupyter_start><jupyter_text>KFP SDK: Component I/O and passing data between pipeline componentsIn this notebook, we'll build an example that shows how data can be passed between pipeline steps.  SetupBefore you run this notebook, ensure that your Google Cloud user account and project are granted access to the Managed Pipelines Experimental. To be granted access to the Managed Pipelines Experimental, fill out this [form](http://go/cloud-mlpipelines-signup) and let your account representative know you have requested access. This notebook is intended to be run on either one of:* [AI Platform Notebooks](https://cloud.google.com/ai-platform-notebooks). See the "AI Platform Notebooks" section in the Experimental [User Guide](https://docs.google.com/document/d/1JXtowHwppgyghnj1N1CT73hwD1caKtWkLcm2_0qGBoI/edit?usp=sharing) for more detail on creating a notebook server instance.* [Google Colab](https://colab.research.google.com/notebooks/intro.ipynb)**To run this notebook on AI Platform Notebooks**, click on the **File** menu, then select "Download .ipynb".  Then, upload that notebook from your local machine to AI Platform Notebooks. (In the AI Platform Notebooks left panel, look for an icon of an arrow pointing up, to upload). We'll first install some libraries and set up some variables.Set `gcloud` to use your project.  **Edit the following cell before running it**.<jupyter_code>PROJECT_ID = 'your-project-id'  # <---CHANGE THIS
!gcloud config set project {PROJECT_ID}<jupyter_output><empty_output><jupyter_text>If you're running this notebook on colab, authenticate with your user account:<jupyter_code>import sys
if 'google.colab' in sys.modules:
  from google.colab import auth
  auth.authenticate_user()<jupyter_output><empty_output><jupyter_text>-----------------**If you're on AI Platform Notebooks**, authenticate with Google Cloud before running the next section, by running```shgcloud auth login```**in the Terminal window** (which you can open via **File** > **New** in the menu). You only need to do this once per notebook instance.  Install the KFP SDK and AI Platform Pipelines client libraryFor Managed Pipelines Experimental, you'll need to download a special version of the AI Platform client library.<jupyter_code>!gsutil cp gs://cloud-aiplatform-pipelines/releases/20210304/aiplatform_pipelines_client-0.1.0.caip20210304-py3-none-any.whl .<jupyter_output><empty_output><jupyter_text>Then, install the libraries and restart the kernel.<jupyter_code>if 'google.colab' in sys.modules:
  USER_FLAG = ''
else:
  USER_FLAG = '--user'
!python3 -m pip install {USER_FLAG} kfp==1.4 --upgrade
!python3 -m pip install {USER_FLAG} aiplatform_pipelines_client-0.1.0.caip20210304-py3-none-any.whl --upgrade
if not 'google.colab' in sys.modules:
  # Automatically restart kernel after installs
  import IPython
  app = IPython.Application.instance()
  app.kernel.do_shutdown(True)<jupyter_output><empty_output><jupyter_text>The KFP version should be >= 1.4.<jupyter_code># Check the KFP version
!python3 -c "import kfp; print('KFP version: {}'.format(kfp.__version__))"<jupyter_output><empty_output><jupyter_text>Set some variables**Before you run the next cell**, **edit it** to set variables for your project.  See the "Before you begin" section of the User Guide for information on creating your API key.  For `BUCKET_NAME`, enter the name of a Cloud Storage (GCS) bucket in your project.  Don't include the `gs://` prefix.<jupyter_code>PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin

# Required Parameters
USER = 'YOUR_USER_NAME' # <---CHANGE THIS
BUCKET_NAME = 'YOUR_BUCKET_NAME'  # <---CHANGE THIS
PIPELINE_ROOT = 'gs://{}/pipeline_root/{}'.format(BUCKET_NAME, USER)

PROJECT_ID = 'YOUR_PROJECT_ID'  # <---CHANGE THIS
REGION = 'us-central1'
API_KEY = 'YOUR_API_KEY'  # <---CHANGE THIS

print('PIPELINE_ROOT: {}'.format(PIPELINE_ROOT))<jupyter_output><empty_output><jupyter_text>Simple two-step pipeline with 'producer' and 'consumer' stepsNow we're ready to define and run a pipeline.  We'll build an example that shows how data can be passed between pipeline steps.We'll first do some imports:<jupyter_code>import time
from kfp import components
from kfp.v2 import compiler
from kfp.v2 import dsl<jupyter_output><empty_output><jupyter_text>Next, we'll define the pipeline components.  We'll do this via the `load_component_from_text` method, which expects a string in `yaml` syntax. Both components use as their base container image the `google/cloud-sdk` image, and run a series of shell commands.  Producer componentThe **Producer** component takes as inputs an `inputValue` parameter (`input_text`, of type `String`), and as output an `outputPath` parameter (`output_value`) and an `outputURI` parameter (`output_artifact`).When the op runs, a Cloud Storage (GCS) path will be generated automatically for `output_value`, which for this component is the second argument passed to the shell command (`$1`). Similarly, a GCS path will be generated for the `output_artifact` (`$2`).When the component op runs, the input values are echoed, and then the `input_text` value is copied via `gsutil` to those paths, making those outputs available for consumption by other components.<jupyter_code>producer_op = components.load_component_from_text("""
name: Producer
inputs:
- {name: input_text, type: String, description: 'Represents an input parameter.'}
outputs:
- {name: output_value, type: String, description: 'Represents an output paramter.'}
- {name: output_artifact, description: 'Represents an output artifact.'}
implementation:
  container:
    image: google/cloud-sdk:latest
    command:
    - sh
    - -c
    - |
      set -e -x
      echo "$0, this is an output parameter" | gsutil cp - "$1"
      echo "$0, this is an output artifact" | gsutil cp - "$2"
    - {inputValue: input_text}
    - {outputPath: output_value}
    - {outputUri: output_artifact}
""")<jupyter_output><empty_output><jupyter_text>Consumer componentThe **Consumer** component has two inputs. When the component op runs, the input values are echoed.  We'll define our pipeline so that the `input_value` and `input_artifact` of the Consumer step are obtained from the outputs of the **Producer** step.<jupyter_code>consumer_op = components.load_component_from_text("""
name: Consumer
inputs:
- {name: input_value, type: String, description: 'Represents an input parameter. It connects to an upstream output parameter.'}
- {name: input_artifact, description: 'Represents an input artifact. It connects to an upstream output artifact.'}
implementation:
  container:
    image: google/cloud-sdk:latest
    command:
    - sh
    - -c
    - |
      set -e -x
      echo "Read from an input parameter: " && echo "$0"
      echo "Read from an input artifact: " && gsutil cat "$1"
    - {inputValue: input_value}
    - {inputUri: input_artifact}
""")<jupyter_output><empty_output><jupyter_text>Define a pipeline using the components and submit a runNext, we'll define a two-step pipeline that uses the producer and consumer ops.  We're setting the Consumer op's `input_value` arg to the output of the Producer op; specifically, the `output_value` output.  Under the hood, this value is obtained by automatically reading the GCS file path to which the Producer step wrote.<jupyter_code>@dsl.pipeline(name='simple-two-step-pipeline-{}-{}'.format(USER, str(int(time.time()))))
def two_step_pipeline(
    text = 'Hello world'
):
  producer = producer_op(input_text=text)
  consumer = consumer_op(input_value=producer.outputs['output_value'], 
                         input_artifact=producer.outputs['output_artifact'])<jupyter_output><empty_output><jupyter_text>Compile the pipeline:<jupyter_code>compiler.Compiler().compile(pipeline_func=two_step_pipeline, 
                            pipeline_root=PIPELINE_ROOT,
                            output_path='two_step_pipeline_job.json')<jupyter_output><empty_output><jupyter_text>Submit the pipeline jobHere, we'll create an API client using the API key you generated.Then, we'll submit the pipeline job by passing the compiled spec to the `create_run_from_job_spec()` method. Note that we're passing a `parameter_values` dict that specifies the pipeline input parameters we want to use.<jupyter_code>from aiplatform.pipelines import client

api_client = client.Client(project_id=PROJECT_ID, region=REGION, api_key=API_KEY)

response = api_client.create_run_from_job_spec(
    job_spec_path='two_step_pipeline_job.json',
    # pipeline_root=PIPELINE_ROOT,  # optional- use if want to override compile-time value
    parameter_values={'text': 'This is some input text'})<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09897097930396577, 'fraction_numerical': 0.006705977569661232, 'mean_word_length': 5.044723969252271, 'pattern_counts': {'":': 0, '<': 54, '<?xml version=': 0, '>': 51, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '124571', 'license': 'Apache-2.0', 'n_tokens_mistral': 2642, 'n_tokens_neox': 2552, 'n_words': 906, 'path': '03-kfp-intro/mp_kfp_producer_consumer.ipynb', 'repo_name': 'jarokaz/caipp', 'size': '8649'}