<jupyter_start><jupyter_text>Dataproc - Submit SparkSQL Job Intended UseA Kubeflow Pipeline component to submit a SparkSql job on YARN in 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.queries | Required. The queries to execute. You do not need to terminate a query with a semicolon. Multiple queries can be specified in one string by separating each with a semicolon. query_file_uri | The HCFS URI of the script that contains SQL queries.script_variables | Optional. Mapping of query variable names to values (equivalent to the Spark SQL command: SET name="value";).sparksql_job | Optional. The full payload of a [SparkSqlJob](https://cloud.google.com/dataproc/docs/reference/rest/v1/SparkSqlJob)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.Directly put your Hive queries in the `queries` list or upload your Hive queries into a file to a Google Cloud Storage (GCS) bucket and place the path in `query_file_uri`. In this sample, we will use a hard coded query in the `queries` list to select data from a public CSV file from GCS. Prepare SparkSQL jobDirectly put your SparkSQL queries in the `queires` list or upload your SparkSQL queries into a file to a Google Cloud Storage (GCS) bucket and place the path in `query_file_uri`. In this sample, we will use a hard coded query in the `queries` list to select data from a public CSV file from GCS.For more details about Spark SQL, please checkout the [programming guide](https://spark.apache.org/docs/latest/sql-programming-guide.html)  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'
QUERY = '''
DROP TABLE IF EXISTS natality_csv;
CREATE EXTERNAL TABLE natality_csv (
  source_year BIGINT, year BIGINT, month BIGINT, day BIGINT, wday BIGINT,
  state STRING, is_male BOOLEAN, child_race BIGINT, weight_pounds FLOAT,
  plurality BIGINT, apgar_1min BIGINT, apgar_5min BIGINT,
  mother_residence_state STRING, mother_race BIGINT, mother_age BIGINT,
  gestation_weeks BIGINT, lmp STRING, mother_married BOOLEAN,
  mother_birth_state STRING, cigarette_use BOOLEAN, cigarettes_per_day BIGINT,
  alcohol_use BOOLEAN, drinks_per_week BIGINT, weight_gain_pounds BIGINT,
  born_alive_alive BIGINT, born_alive_dead BIGINT, born_dead BIGINT,
  ever_born BIGINT, father_race BIGINT, father_age BIGINT,
  record_weight BIGINT
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 'gs://public-datasets/natality/csv';

SELECT * FROM natality_csv LIMIT 10;'''
EXPERIMENT_NAME = 'Dataproc - Submit SparkSQL Job'
COMPONENT_SPEC_URI = 'https://raw.githubusercontent.com/kubeflow/pipelines/d2f5cc92a46012b9927209e2aaccab70961582dc/components/gcp/dataproc/submit_sparksql_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_sparksql_job_op = comp.load_component_from_url(COMPONENT_SPEC_URI)
display(dataproc_submit_sparksql_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 SparkSQL job pipeline',
    description='Dataproc submit SparkSQL job pipeline'
)
def dataproc_submit_sparksql_job_pipeline(
    project_id = PROJECT_ID, 
    region = REGION,
    cluster_name = CLUSTER_NAME,
    queries = json.dumps([QUERY]),
    query_file_uri = '',
    script_variables = '', 
    sparksql_job='', 
    job='', 
    wait_interval='30'
):
    dataproc_submit_sparksql_job_op(project_id, region, cluster_name, queries, query_file_uri,
        script_variables, sparksql_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_sparksql_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.08038932948810382, 'fraction_numerical': 0.007390050468637346, 'mean_word_length': 5.693606755126659, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 6, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '87926', 'license': 'Apache-2.0', 'n_tokens_mistral': 1803, 'n_tokens_neox': 1719, 'n_words': 538, 'path': 'components/gcp/dataproc/submit_sparksql_job/sample.ipynb', 'repo_name': 'thedriftofwords/pipelines', 'size': '5548'}