<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Import the sample data into BigQuery and Datastore
This notebook is the first of two notebooks that guide you through completing the prerequisites for running the Real-time Item-to-item Recommendation with BigQuery ML Matrix Factorization and ScaNN solution.
Use this notebook to complete the following tasks
Step1:  Import libraries
Step2:  Configure GCP environment settings
Update the following variables to reflect the values for your GCP environment
Step3:  Authenticate your GCP account
This is required if you run the notebook in Colab. If you use an AI Platform notebook, you should already be authenticated.
Step4:  Copy the public playlist data into your BigQuery dataset
Create the BigQuery dataset
Step5:  Define the Dataflow pipeline
The pipeline selects songs where the track_data_title field isn't NULL and the track_data_id field is greater than 0.
Step6:  Run the Dataflow pipeline
This pipeline takes approximately 15 minutes to run.
Step7:  Create the vw_item_groups view
Create the recommendations.vw_item_groups view to focus on song and playlist data.
To adapt this view to your own data, you would need to map your item identifier, for example product SKU, to item_Id, and your context identifier, for example purchase order number, to group_Id.
Step8:  Export song information to Datastore
Export data from the track_title and artist fields to Datastore.
Define the Dataflow pipeline
Step9:  Run the Dataflow pipeline
This pipeline takes approximately 15 minutes to run.
<ASSISTANT_TASK:>
Python Code:
!pip install -q -U apache-beam[gcp]
# Automatically restart kernel after installs
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)  
Explanation: Import the sample data into BigQuery and Datastore
This notebook is the first of two notebooks that guide you through completing the prerequisites for running the Real-time Item-to-item Recommendation with BigQuery ML Matrix Factorization and ScaNN solution.
Use this notebook to complete the following tasks:
Importing the playlist table from the public BigQuery dataset to your BigQuery dataset.
Creating the vw_item_groups view that contains the item data used to compute item co-occurence.
Exporting song title and artist information to Datastore to make it available for lookup when making similar song recommendations. 
Before starting this notebook, you must set up the GCP environment.
After completing this notebook, run the 00_prep_bq_procedures notebook to complete the solution prerequisites.
Setup
Install the required Python packages, configure the environment variables, and authenticate your GCP account.
End of explanation
import os
from datetime import datetime
import apache_beam as beam
from apache_beam.io.gcp.datastore.v1new.datastoreio import WriteToDatastore
Explanation: Import libraries
End of explanation
PROJECT_ID = 'yourProject' # Change to your project.
BUCKET = 'yourBucketName' # Change to the bucket you created.
BQ_REGION = 'yourBigQueryRegion' # Change to your BigQuery region.
DF_REGION = 'yourDataflowRegion' # Change to your Dataflow region.
BQ_DATASET_NAME = 'recommendations'
BQ_TABLE_NAME = 'playlist'
DS_KIND = 'song'
!gcloud config set project $PROJECT_ID
Explanation: Configure GCP environment settings
Update the following variables to reflect the values for your GCP environment:
PROJECT_ID: The ID of the Google Cloud project you are using to implement this solution.
BUCKET: The name of the Cloud Storage bucket you created to use with this solution. The BUCKET value should be just the bucket name, so myBucket rather than gs://myBucket.
BQ_REGION: The region to use for the BigQuery dataset.
DF_REGION: The region to use for the Dataflow job. Choose the same region that you used for the BQ_REGION variable to avoid issues around reading/writing in different locations.
End of explanation
try:
  from google.colab import auth
  auth.authenticate_user()
  print("Colab user is authenticated.")
except: pass
Explanation: Authenticate your GCP account
This is required if you run the notebook in Colab. If you use an AI Platform notebook, you should already be authenticated.
End of explanation
!bq mk --dataset \
    --location={BQ_REGION} \
    --project_id={PROJECT_ID} \
    --headless=True \
    {PROJECT_ID}:{BQ_DATASET_NAME}
Explanation: Copy the public playlist data into your BigQuery dataset
Create the BigQuery dataset
End of explanation
def run_copy_bq_data_pipeline(args):
  schema = 'list_Id:INT64, track_Id:INT64, track_title:STRING, track_artist:STRING'
  query = '''
    SELECT 
      id list_Id, 
      tracks_data_id track_Id, 
      tracks_data_title track_title,
      tracks_data_artist_name track_artist
    FROM `bigquery-samples.playlists.playlist`
    WHERE tracks_data_title IS NOT NULL AND tracks_data_id > 0
    GROUP BY list_Id, track_Id, track_title, track_artist;
  '''
  pipeline_options = beam.options.pipeline_options.PipelineOptions(**args)
  with beam.Pipeline(options=pipeline_options) as pipeline:
    _ = (
        pipeline
        | 'ReadFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(
            project=PROJECT_ID, query=query, use_standard_sql=True))
        | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
            table=BQ_TABLE_NAME, dataset=BQ_DATASET_NAME, project=PROJECT_ID,
            schema=schema, 
            create_disposition='CREATE_IF_NEEDED',
            write_disposition='WRITE_TRUNCATE'
        )
    )
Explanation: Define the Dataflow pipeline
The pipeline selects songs where the track_data_title field isn't NULL and the track_data_id field is greater than 0.
End of explanation
DATASET = 'playlist'
RUNNER = 'DataflowRunner'
job_name = f'copy-bigquery-{datetime.utcnow().strftime("%y%m%d%H%M%S")}'
args = {
    'job_name': job_name,
    'runner': RUNNER,
    'project': PROJECT_ID,
    'temp_location': f'gs://{BUCKET}/dataflow_tmp',
    'region': DF_REGION
}
print("Pipeline args are set.")
print("Running pipeline...")
%time run_copy_bq_data_pipeline(args)
print("Pipeline is done.")
Explanation: Run the Dataflow pipeline
This pipeline takes approximately 15 minutes to run.
End of explanation
%%bigquery  --project $PROJECT_ID
CREATE OR REPLACE VIEW `recommendations.vw_item_groups`
AS
SELECT
  list_Id AS group_Id,
  track_Id AS item_Id
FROM  
  `recommendations.playlist` 
Explanation: Create the vw_item_groups view
Create the recommendations.vw_item_groups view to focus on song and playlist data.
To adapt this view to your own data, you would need to map your item identifier, for example product SKU, to item_Id, and your context identifier, for example purchase order number, to group_Id.
End of explanation
def create_entity(song_info, kind):
  from apache_beam.io.gcp.datastore.v1new.types import Entity
  from apache_beam.io.gcp.datastore.v1new.types import Key
  track_Id = song_info.pop("track_Id")
  key = Key([kind, track_Id])
  song_entity = Entity(key)
  song_entity.set_properties(song_info)
  return song_entity
def run_export_to_datatore_pipeline(args):
    query = f'''
      SELECT  
        track_Id, 
        MAX(track_title) track_title, 
        MAX(track_artist) artist
      FROM 
        `{BQ_DATASET_NAME}.{BQ_TABLE_NAME}`
      GROUP BY track_Id
    '''
    pipeline_options = beam.options.pipeline_options.PipelineOptions(**args)
    with beam.Pipeline(options=pipeline_options) as pipeline:
      _ = (
        pipeline
        | 'ReadFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(
            project=PROJECT_ID, query=query, use_standard_sql=True))
        | 'ConvertToDatastoreEntity' >> beam.Map(create_entity, DS_KIND)
        | 'WriteToDatastore' >> WriteToDatastore(project=PROJECT_ID)
      )
Explanation: Export song information to Datastore
Export data from the track_title and artist fields to Datastore.
Define the Dataflow pipeline
End of explanation
import os
from datetime import datetime
DATASET = 'playlist'
RUNNER = 'DataflowRunner'
job_name = f'load-datastore-{datetime.utcnow().strftime("%y%m%d%H%M%S")}'
args = {
    'job_name': job_name,
    'runner': RUNNER,
    'project': PROJECT_ID,
    'temp_location': f'gs://{BUCKET}/dataflow_tmp',
    'region': DF_REGION
}
print("Pipeline args are set.")
print("Running pipeline...")
%time run_export_to_datatore_pipeline(args)
print("Pipeline is done.")
Explanation: Run the Dataflow pipeline
This pipeline takes approximately 15 minutes to run.
End of explanation
<END_TASK>