<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Name
Gather training data by querying BigQuery 
Labels
GCP, BigQuery, Kubeflow, Pipeline
Summary
A Kubeflow Pipeline component to submit a query to BigQuery and store the result in a Cloud Storage bucket.
Details
Intended use
Use this Kubeflow component to
Step1:  Load the component using KFP SDK
Step2:  Sample
Note
Step3:  Set sample parameters
Step4:  Run the component as a single pipeline
Step5:  Compile the pipeline
Step6:  Submit the pipeline for execution
Step7:  Inspect the output
<ASSISTANT_TASK:>
Python Code:
%%capture --no-stderr
!pip3 install kfp --upgrade
Explanation: Name
Gather training data by querying BigQuery 
Labels
GCP, BigQuery, Kubeflow, Pipeline
Summary
A Kubeflow Pipeline component to submit a query to BigQuery and store the result in a Cloud Storage bucket.
Details
Intended use
Use this Kubeflow component to:
*   Select training data by submitting a query to BigQuery.
*   Output the training data into a Cloud Storage bucket as CSV files.
Runtime arguments:
| Argument | Description | Optional | Data type | Accepted values | Default |
|----------|-------------|----------|-----------|-----------------|---------|
| query | The query used by BigQuery to fetch the results. | No | String |  |  |
| project_id | The project ID of the Google Cloud Platform (GCP) project to use to execute the query. | No | GCPProjectID |  |  |
| dataset_id | The ID of the persistent BigQuery dataset to store the results of the query. If the dataset does not exist, the operation will create a new one. | Yes | String |  | None |
| table_id | The ID of the BigQuery table to store the results of the query. If the table ID is absent, the operation will generate a random ID for the table. | Yes | String |  | None |
| output_gcs_path | The path to the Cloud Storage bucket to store the query output. | Yes | GCSPath |  | None |
| dataset_location | The location where the dataset is created. Defaults to US. | Yes | String |  | US |
| job_config | The full configuration specification for the query job. See QueryJobConfig for details. | Yes | Dict | A JSONobject which has the same structure as QueryJobConfig | None |
Input data schema
The input data is a BigQuery job containing a query that pulls data f rom various sources. 
Output:
Name | Description | Type
:--- | :---------- | :---
output_gcs_path | The path to the Cloud Storage bucket containing the query output in CSV format. | GCSPath
Cautions & requirements
To use the component, the following requirements must be met:
The BigQuery API is enabled.
The component can authenticate to use GCP APIs. Refer to Authenticating Pipelines to GCP for details.
The Kubeflow user service account is a member of the roles/bigquery.admin role of the project.
The Kubeflow user service account is a member of the roles/storage.objectCreatorrole of the Cloud Storage output bucket.
Detailed description
This Kubeflow Pipeline component is used to:
*   Submit a query to BigQuery.
    *   The query results are persisted in a dataset table in BigQuery.
    *   An extract job is created in BigQuery to extract the data from the dataset table and output it to a Cloud Storage bucket as CSV files.
Use the code below as an example of how to run your BigQuery job.
Sample
Note: The following sample code works in an IPython notebook or directly in Python code.
Set sample parameters
End of explanation
import kfp.components as comp
bigquery_query_op = comp.load_component_from_url(
    'https://raw.githubusercontent.com/kubeflow/pipelines/01a23ae8672d3b18e88adf3036071496aca3552d/components/gcp/bigquery/query/component.yaml')
help(bigquery_query_op)
Explanation: Load the component using KFP SDK
End of explanation
QUERY = 'SELECT * FROM `bigquery-public-data.stackoverflow.posts_questions` LIMIT 10'
Explanation: Sample
Note: The following sample code works in IPython notebook or directly in Python code.
In this sample, we send a query to get the top questions from stackdriver public data and output the data to a Cloud Storage bucket. Here is the query:
End of explanation
# Required Parameters
PROJECT_ID = '<Please put your project ID here>'
GCS_WORKING_DIR = 'gs://<Please put your GCS path here>' # No ending slash
# Optional Parameters
EXPERIMENT_NAME = 'Bigquery -Query'
OUTPUT_PATH = '{}/bigquery/query/questions.csv'.format(GCS_WORKING_DIR)
Explanation: Set sample parameters
End of explanation
import kfp.dsl as dsl
import json
@dsl.pipeline(
    name='Bigquery query pipeline',
    description='Bigquery query pipeline'
)
def pipeline(
    query=QUERY, 
    project_id = PROJECT_ID, 
    dataset_id='', 
    table_id='', 
    output_gcs_path=OUTPUT_PATH, 
    dataset_location='US', 
    job_config=''
):
    bigquery_query_op(
        query=query, 
        project_id=project_id, 
        dataset_id=dataset_id, 
        table_id=table_id, 
        output_gcs_path=output_gcs_path, 
        dataset_location=dataset_location, 
        job_config=job_config)
Explanation: Run the component as a single pipeline
End of explanation
pipeline_func = pipeline
pipeline_filename = pipeline_func.__name__ + '.zip'
import kfp.compiler as compiler
compiler.Compiler().compile(pipeline_func, pipeline_filename)
Explanation: Compile the pipeline
End of explanation
#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)
Explanation: Submit the pipeline for execution
End of explanation
!gsutil cat $OUTPUT_PATH
Explanation: Inspect the output
End of explanation
<END_TASK>