<jupyter_start><jupyter_text>Dataproc - Submit PySpark Job Intended UseA Kubeflow Pipeline component to submit a PySpark job to Google Cloud Dataproc service.  Run-Time Parameters:Name | Description:--- | :----------project_id | Required. The ID of the Google Cloud Platform project that the cluster belongs to.region | Required. The Cloud Dataproc region in which to handle the request.cluster_name | Required. The cluster to run the job.main_python_file_uri | Required. The HCFS URI of the main Python file to use as the driver. Must be a .py file.args | Optional. The arguments to pass to the driver. Do not include arguments, such as --conf, that can be set as job properties, since a collision may occur that causes an incorrect job submission.pyspark_job | Optional. The full payload of a [PySparkJob](https://cloud.google.com/dataproc/docs/reference/rest/v1/PySparkJob).job | Optional. The full payload of a [Dataproc job](https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.jobs).wait_interval | Optional. The wait seconds between polling the operation. Defaults to 30s. Output:Name | Description:--- | :----------job_id | The ID of the created job.  SampleNote: the sample code below works in both IPython notebook or python code directly. Setup a Dataproc clusterFollow the [guide](https://cloud.google.com/dataproc/docs/guides/create-cluster) to create a new Dataproc cluster or reuse an existing one. Prepare PySpark jobUpload your PySpark code file to a Google Cloud Storage (GCS) bucket. For example, here is a public accessible hello-world.py in GCS:<jupyter_code>!gsutil cat gs://dataproc-examples-2f10d78d114f6aaec76462e3c310f31f/src/pyspark/hello-world/hello-world.py<jupyter_output><empty_output><jupyter_text>Set sample parameters<jupyter_code>PROJECT_ID = '<Please put your project ID here>'
CLUSTER_NAME = '<Please put your existing cluster name here>'
REGION = 'us-central1'
PYSPARK_FILE_URI = 'gs://dataproc-examples-2f10d78d114f6aaec76462e3c310f31f/src/pyspark/hello-world/hello-world.py'
ARGS = ''
EXPERIMENT_NAME = 'Dataproc - Submit PySpark Job'
SUBMIT_PYSPARK_JOB_SPEC_URI = 'https://raw.githubusercontent.com/kubeflow/pipelines/d2f5cc92a46012b9927209e2aaccab70961582dc/components/gcp/dataproc/submit_pyspark_job/component.yaml'<jupyter_output><empty_output><jupyter_text>Install KFP SDKInstall the SDK (Uncomment the code if the SDK is not installed before)<jupyter_code># KFP_PACKAGE = 'https://storage.googleapis.com/ml-pipeline/release/0.1.12/kfp.tar.gz'
# !pip3 install $KFP_PACKAGE --upgrade<jupyter_output><empty_output><jupyter_text>Load component definitions<jupyter_code>import kfp.components as comp

dataproc_submit_pyspark_job_op = comp.load_component_from_url(SUBMIT_PYSPARK_JOB_SPEC_URI)
display(dataproc_submit_pyspark_job_op)<jupyter_output><empty_output><jupyter_text>Here is an illustrative pipeline that uses the component<jupyter_code>import kfp.dsl as dsl
import kfp.gcp as gcp
import json
@dsl.pipeline(
    name='Dataproc submit PySpark job pipeline',
    description='Dataproc submit PySpark job pipeline'
)
def dataproc_submit_pyspark_job_pipeline(
    project_id = PROJECT_ID, 
    region = REGION,
    cluster_name = CLUSTER_NAME,
    main_python_file_uri = PYSPARK_FILE_URI, 
    args = ARGS, 
    pyspark_job='{}', 
    job='{}', 
    wait_interval='30'
):
    dataproc_submit_pyspark_job_op(project_id, region, cluster_name, main_python_file_uri, 
        args, pyspark_job, job, wait_interval).apply(gcp.use_gcp_secret('user-gcp-sa'))<jupyter_output><empty_output><jupyter_text>Compile the pipeline<jupyter_code>pipeline_func = dataproc_submit_pyspark_job_pipeline
pipeline_filename = pipeline_func.__name__ + '.pipeline.tar.gz'
import kfp.compiler as compiler
compiler.Compiler().compile(pipeline_func, pipeline_filename)<jupyter_output><empty_output><jupyter_text>Submit the pipeline for execution<jupyter_code>#Specify pipeline argument values
arguments = {}

#Get or create an experiment and submit a pipeline run
import kfp
client = kfp.Client()
experiment = client.create_experiment(EXPERIMENT_NAME)

#Submit a pipeline run
run_name = pipeline_func.__name__ + ' run'
run_result = client.run_pipeline(experiment.id, run_name, pipeline_filename, arguments)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0912707699508542, 'fraction_numerical': 0.01802012637491224, 'mean_word_length': 5.871382636655949, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '271313', 'license': 'Apache-2.0', 'n_tokens_mistral': 1459, 'n_tokens_neox': 1401, 'n_words': 354, 'path': 'components/gcp/dataproc/submit_pyspark_job/sample.ipynb', 'repo_name': 'thedriftofwords/pipelines', 'size': '4273'}