project = 'Input your PROJECT ID'
region = 'Input GCP region' # For example, 'us-central1'
output = 'Input your GCS bucket name' # No ending slash

"""
Explanation: GCP Dataflow Component Sample
A Kubeflow Pipeline component that prepares data by submitting an Apache Beam job (authored in Python) to Cloud Dataflow for execution. The Python Beam code is run with Cloud Dataflow Runner.
Intended use
Use this component to run a Python Beam code to submit a Cloud Dataflow job as a step of a Kubeflow pipeline. 
Runtime arguments
Name | Description | Optional |  Data type| Accepted values | Default |
:--- | :----------| :----------| :----------| :----------| :---------- |
python_file_path |  The path to the Cloud Storage bucket or local directory containing the Python file to be run. |  |  GCSPath |  |  |
project_id |  The ID of the Google Cloud Platform (GCP) project  containing the Cloud Dataflow job.| | String | | |
region |  The Google Cloud Platform (GCP) region to run the Cloud Dataflow job.| | String | | |
staging_dir  |   The path to the Cloud Storage directory where the staging files are stored. A random subdirectory will be created under the staging directory to keep the  job information.This is done so that you can resume the job in case of failure. staging_dir is passed as the command line arguments (staging_location and temp_location) of the Beam code. |   Yes  |   GCSPath  |   |   None  |
requirements_file_path |   The path to the Cloud Storage bucket or local directory containing the pip requirements file. | Yes | GCSPath |  | None |
args |  The list of arguments to pass to the Python file. | No |  List | A list of string arguments | None |
wait_interval |  The number of seconds to wait between calls to get the status of the job. | Yes | Integer  |  | 30 |
Input data schema
Before you use the component, the following files must be ready in a Cloud Storage bucket:
- A Beam Python code file.
- A  requirements.txt file which includes a list of dependent packages.
The Beam Python code should follow the Beam programming guide as well as the following additional requirements to be compatible with this component:
- It accepts the command line arguments --project, --region, --temp_location, --staging_location, which are standard Dataflow Runner options.
- It enables info logging before the start of a Cloud Dataflow job in the Python code. This is important to allow the component to track the status and ID of the job that is created. For example, calling logging.getLogger().setLevel(logging.INFO) before any other code.
Output
Name | Description
:--- | :----------
job_id | The id of the Cloud Dataflow job that is created.
Cautions & requirements
To use the components, the following requirements must be met:
- Cloud Dataflow API is enabled.
- The component is running under a secret Kubeflow user service account in a Kubeflow Pipeline cluster.  For example:
component_op(...)
The Kubeflow user service account is a member of:
- roles/dataflow.developer role of the project.
- roles/storage.objectViewer role of the Cloud Storage Objects python_file_path and requirements_file_path.
- roles/storage.objectCreator role of the Cloud Storage Object staging_dir. 
Detailed description
The component does several things during the execution:
- Downloads python_file_path and requirements_file_path to local files.
- Starts a subprocess to launch the Python program.
- Monitors the logs produced from the subprocess to extract the Cloud Dataflow job information.
- Stores the Cloud Dataflow job information in staging_dir so the job can be resumed in case of failure.
- Waits for the job to finish.
Setup
End of explanation
"""


!python3 -m pip install 'kfp>=0.1.31' --quiet

"""
Explanation: Install Pipeline SDK
End of explanation
"""


import kfp.deprecated.components as comp

dataflow_python_op = comp.load_component_from_url(
    'https://raw.githubusercontent.com/kubeflow/pipelines/1.7.0-rc.3/components/gcp/dataflow/launch_python/component.yaml')

help(dataflow_python_op)

"""
Explanation: Load the component using KFP SDK
End of explanation
"""


!gsutil cat gs://ml-pipeline/sample-pipeline/word-count/wc.py

"""
Explanation: Use the wordcount python sample
In this sample, we run a wordcount sample code in a Kubeflow Pipeline. The output will be stored in a Cloud Storage bucket. Here is the sample code:
End of explanation
"""


import kfp.deprecated as kfp
from kfp.deprecated import dsl, Client
import json
@dsl.pipeline(
    name='dataflow-launch-python-pipeline',
    description='Dataflow launch python pipeline'
)
def pipeline(
    python_file_path = 'gs://ml-pipeline/sample-pipeline/word-count/wc.py',
    project_id = project,
    region = region,
    staging_dir = output,
    requirements_file_path = 'gs://ml-pipeline/sample-pipeline/word-count/requirements.txt',
    wait_interval = 30
):
    dataflow_python_op(
        python_file_path = python_file_path, 
        project_id = project_id, 
        region = region, 
        staging_dir = staging_dir, 
        requirements_file_path = requirements_file_path, 
        args = json.dumps(['--output', f'{staging_dir}/wc/wordcount.out']),
        wait_interval = wait_interval)

"""
Explanation: Example pipeline that uses the component
End of explanation
"""


Client().create_run_from_pipeline_func(pipeline, arguments={})

"""
Explanation: Submit the pipeline for execution
End of explanation
"""


!gsutil cat $output/wc/wordcount.out

"""
Explanation: Inspect the output
End of explanation
"""


