<jupyter_start><jupyter_text>OverviewThis notebook does the following:* Demonstrates how you can connect Amazon SageMaker Studio Sparkmagic kernel to a kerberized EMR cluster* Explore and query data from a Hive table * Data pre-processing and feature engineering using SageMaker Processing job* Trains a model using blazingtext algorithm * Deploys the model to a SageMaker Hosting Endpoint* Get Predictions from the Endpoint----------When using PySpark kernel notebooks, there is no need to create a SparkContext or a HiveContext; those are all created for you automatically when you run the first code cell, and you'll be able to see the progress printed. The contexts are created with the following variable names:- SparkContext (sc)- HiveContext (sqlContext)---------- PySpark magics The PySpark kernel provides some predefined “magics”, which are special commands that you can call with `%%` (e.g. `%%MAGIC` ). The magic command must be the first word in a code cell and allow for multiple lines of content. You can’t put comments before a cell magic.For more information on magics, see [here](http://ipython.readthedocs.org/en/stable/interactive/magics.html). Running locally (%%local)You can use the `%%local` magic to run your code locally on the Jupyter server without going to Spark. When you use %%local all subsequent lines in the cell will be executed locally. The code in the cell must be valid Python code.  Explore and query data Update the EMR Cluster IDReplace "x-xxxxxxxxxxx" in the next cell with the  ID of the EMR cluster that was deployed as part of this solution. You can find this information on the [EMR console](https://console.aws.amazon.com/elasticmapreduce/home) under the "ID" field<jupyter_code>%%local 
emr_cluster_id="x-xxxxxxxxxxx"<jupyter_output><empty_output><jupyter_text>Next, we will connect to the EMR cluster as "user1" using the "sm-sparkmagic" CLI. If you dont set the user name, jobs will apear as livy user<jupyter_code>%%local
!sm-sparkmagic connect --cluster-id {emr_cluster_id} --user-name "user1"<jupyter_output><empty_output><jupyter_text>Before running the next cell, click on the launch terminal in current SageMaker image icon ($_) in the toolbar above and access the terminal. In the terminal window, type kinit user1 and when prompted for a password type in the password for "<PASSWORD>"*Note: First time, use "<PASSWORD>" for the password. You will receive a message saying that your password is expired and you must change it now. Enter a new password when prompted.*```Password for user1@EC2.INTERNAL: Password expired.  You must change it now.Enter new password: Enter it again: ```**Before proceeding, restart the kernel using the restart kernel icon from the toolbar. This is required so that SparkMagic can pickup the generated kerberos configuration**.  Session information (%%info)Livy is an open source REST server for Spark. When you execute a code cell in a sparkmagic notebook, it creates a Livy session to execute your code. `%%info` magic will display the current Livy session information.<jupyter_code>%%info<jupyter_output><empty_output><jupyter_text>You can configure the session creation parameters using the %%configure magic if needed<jupyter_code>%%configure -f 
{"driverMemory": "1000M", "executorCores": 2}<jupyter_output><empty_output><jupyter_text>In the next cell, we will use the HiveContext to query Hive and look at the databases and tables<jupyter_code>sqlContext = HiveContext(sqlContext) 

dbs = sqlContext.sql("show databases")
dbs.show()

tables = sqlContext.sql("show tables")
tables.show()<jupyter_output><empty_output><jupyter_text>Next, we will query the movie_reviews table and get the data into a spark dataframe. You can visualize the data from the remote cluster locally in the notebook<jupyter_code>movie_reviews = sqlContext.sql("select * from movie_reviews").cache()<jupyter_output><empty_output><jupyter_text>Let's look at the data size and size of each class (positive and negative) and visualize it. You can see that we have a balanced dataset with equal number on both classes (25000 each)<jupyter_code># Shape
print((movie_reviews.count(), len(movie_reviews.columns)))
# count of both positive and negative sentiments
movie_reviews.groupBy('sentiment').count().show()

pos_reviews = movie_reviews.filter(movie_reviews.sentiment == 'positive').collect()
neg_reviews = movie_reviews.filter(movie_reviews.sentiment == 'negative').collect()
import matplotlib.pyplot as plt
def plot_counts(positive,negative):
    plt.rcParams['figure.figsize']=(6,6)
    plt.bar(0,positive,width=0.6,label='Positive Reviews',color='Green')
    plt.bar(2,negative,width=0.6,label='Negative Reviews',color='Red')
    handles, labels = plt.gca().get_legend_handles_labels()
    by_label = dict(zip(labels, handles))
    plt.legend(by_label.values(), by_label.keys())
    plt.ylabel('Count')
    plt.xlabel('Type of Review')
    plt.tick_params(
        axis='x',          
        which='both',      
        bottom=False,      
        top=False,         
        labelbottom=False) 
    plt.show()
    
plot_counts(len(pos_reviews),len(neg_reviews))
%matplot plt<jupyter_output><empty_output><jupyter_text>Next, Let's inspect length of reviews using the pyspark.sql.functions module<jupyter_code>from pyspark.sql.functions import length
reviewlengthDF = movie_reviews.select(length('review').alias('Length of Review')) 
reviewlengthDF.show()<jupyter_output><empty_output><jupyter_text>You can also execute SparkSQL queries using the %%sql magic and save results to a local data frame. This allows for quick data exploration. Max rows returned by default is 2500. You can set the max rows by using the -n argument.<jupyter_code>%%sql -o movie_reviews_sparksql_df -n 10
select * from movie_reviews<jupyter_output><empty_output><jupyter_text>You can access and explore the data in the dataframe locally<jupyter_code>%%local 
movie_reviews_sparksql_df.head(10)<jupyter_output><empty_output><jupyter_text>Using the SageMaker Studio sparkmagic kernel, you can train machine learning models in the Spark cluster using the *SageMaker Spark library*. SageMaker Spark is an open source Spark library for Amazon SageMaker. For examples, see [here](https://github.com/aws/sagemaker-sparkexample-using-sagemaker-spark-with-any-sagemaker-algorithm)In this notebook however, we will use SageMaker experiments, trial and estimator to train a model and deploy the model using SageMaker realtime endpoint hostingIn the next cell, we will install the necessary libraries<jupyter_code>%%local
import sys
!{sys.executable} -m pip install -U  "sagemaker>=1.72.1,<2.0.0"  
!{sys.executable} -m pip install sagemaker-experiments 
!{sys.executable} -m pip show sagemaker<jupyter_output><empty_output><jupyter_text>Next, we will import libraries and set global definitions<jupyter_code>%%local
import sagemaker
import boto3
import botocore
from botocore.exceptions import ClientError
from time import strftime, gmtime
import json
from sagemaker import get_execution_role

from smexperiments.experiment import Experiment
from smexperiments.trial import Trial
%%local 
sess = boto3.Session()
region_name = sess.region_name
role = sagemaker.get_execution_role()
sm_runtime = boto3.Session().client('sagemaker-runtime')<jupyter_output><empty_output><jupyter_text>In the next cell, we will create a new S3 bucket that will be used for storing the training and validation data<jupyter_code>%%local 
stsclient = boto3.client("sts", region_name=region_name)
s3client = boto3.client("s3", region_name=region_name)

aws_account_id = stsclient.get_caller_identity()["Account"] 
bucket = "sagemaker-studio-pyspark-{}-{}".format(region_name, aws_account_id)
key = "sentiment/movie_reviews.csv"
smprocessing_input = "s3://{}/{}".format(bucket, key)

try:
    if region_name=="us-east-1":
           s3client.create_bucket(Bucket=bucket)
    else:
           s3client.create_bucket(Bucket=bucket, CreateBucketConfiguration={
                'LocationConstraint': region_name})
except ClientError as e:
    error_code = e.response['Error']['Code']
    message = e.response['Error']['Message']
    if error_code == 'BucketAlreadyOwnedByYou':
        print ('A bucket with the same name already exists in your account - using the same bucket.')
        pass
    else:
        print("Error->{}:{}".format(error_code, message))<jupyter_output><empty_output><jupyter_text>Send the following variables to spark<jupyter_code>%%send_to_spark -i bucket -t str -n bucket
%%send_to_spark -i key -t str -n key<jupyter_output><empty_output><jupyter_text>Convert the spark dataframe received by querying the hive table (using the sqlContext.sql above) to a pandas dataframe and upload the data to the S3 bucket<jupyter_code>movie_reviews_df = movie_reviews.toPandas()
import boto3
from io import StringIO
csv_buffer = StringIO()
movie_reviews_df.to_csv(csv_buffer)
s3_resource = boto3.resource('s3')
s3_resource.Object(bucket, key).put(Body=csv_buffer.getvalue())<jupyter_output><empty_output><jupyter_text>Pre-process data and feature engineering Amazon SageMaker Processing jobs using the Scikit-learn Processor With Amazon SageMaker Processing jobs, you can leverage a simplified, managed experience to run data pre- or post-processing and model evaluation workloads on the Amazon SageMaker platform.A processing job downloads input from Amazon Simple Storage Service (Amazon S3), then uploads outputs to Amazon S3 during or after the processing job.The cell below shows how to run scikit-learn scripts using a Docker image provided and maintained by SageMaker to preprocess data.*Note: We will use a "ml.m5.xlarge" instance as the instance type for sagemaker processing, training and model hosting. If you don't have access to this instance type and see a "ResourceLimitExceeded" error, use another instance type that you have access to. You can also request a service limit increase using [AWS Support Center](https://console.aws.amazon.com/support)*<jupyter_code>%%local
instance_type_smprocessing="ml.m5.xlarge"
instance_type_smtraining="ml.m5.xlarge"
instance_type_smendpoint="ml.m5.xlarge"
%%local
from sagemaker.sklearn.processing import SKLearnProcessor
sklearn_processor = SKLearnProcessor(framework_version='0.20.0',
                                     role=role,
                                     instance_type=instance_type_smprocessing,
                                     instance_count=1)
%%local
print(smprocessing_input)
from sagemaker.processing import ProcessingInput, ProcessingOutput

sklearn_processor.run(code='preprocessing.py',
                      inputs=[ProcessingInput(
                        source=smprocessing_input,
                        destination='/opt/ml/processing/input')],
                      outputs=[ProcessingOutput(output_name='train_data',
                                                source='/opt/ml/processing/train'),
                               ProcessingOutput(output_name='validation_data',
                                                source='/opt/ml/processing/validation')],
                      arguments=['--train-test-split-ratio', '0.2']
                     )

preprocessing_job_description = sklearn_processor.jobs[-1].describe()

output_config = preprocessing_job_description['ProcessingOutputConfig']
for output in output_config['Outputs']:
    if output['OutputName'] == 'train_data':
        preprocessed_training_data = output['S3Output']['S3Uri']
    if output['OutputName'] == 'validation_data':
        preprocessed_validation_data = output['S3Output']['S3Uri']
%%local 
print(preprocessed_training_data)
print(preprocessed_validation_data)
%%local
prefix = 'blazingtext/supervised' 
s3_train_data = preprocessed_training_data
s3_validation_data = preprocessed_validation_data
s3_output_location = 's3://{}/{}/output'.format(bucket, prefix)<jupyter_output><empty_output><jupyter_text>Train a SageMaker model Amazon SageMaker ExperimentsAmazon SageMaker Experiments allows us to keep track of model training; organize related models together; and log model configuration, parameters, and metrics to reproduce and iterate on previous models and compare models. Let's create the experiment, trial, and train the model. To reduce cost, the training code below uses spot instances.<jupyter_code>%%local
sm_session = sagemaker.session.Session()

create_date = strftime("%Y-%m-%d-%H-%M-%S", gmtime())
sentiment_experiment = Experiment.create(experiment_name="sentimentdetection-{}".format(create_date), 
                                              description="Detect sentiment in text", 
                                              sagemaker_boto_client=boto3.client('sagemaker'))

trial = Trial.create(trial_name="sentiment-trial-blazingtext-{}".format(strftime("%Y-%m-%d-%H-%M-%S", gmtime())), 
                     experiment_name=sentiment_experiment.experiment_name,
                     sagemaker_boto_client=boto3.client('sagemaker'))

container = sagemaker.amazon.amazon_estimator.get_image_uri(region_name, "blazingtext", "latest")
print('Using SageMaker BlazingText container: {} ({})'.format(container, region_name))
%%local 
train_use_spot_instances = True
train_max_run=3600
train_max_wait = 3600 if train_use_spot_instances else None

bt_model = sagemaker.estimator.Estimator(container,
                                         role, 
                                         train_instance_count=1, 
                                         train_instance_type=instance_type_smtraining,
                                         train_volume_size = 30,
                                         input_mode= 'File',
                                         output_path=s3_output_location,
                                         sagemaker_session=sm_session,
                                         train_use_spot_instances=train_use_spot_instances,
                                         train_max_run=train_max_run,
                                         train_max_wait=train_max_wait)
%%local 
bt_model.set_hyperparameters(mode="supervised",
                            epochs=10,
                            min_count=2,
                            learning_rate=0.005328,
                            vector_dim=286,
                            early_stopping=True,
                            patience=4,
                            min_epochs=5,
                            word_ngrams=2)
%%local
train_data = sagemaker.session.s3_input(s3_train_data, distribution='FullyReplicated', 
                        content_type='text/plain', s3_data_type='S3Prefix')
validation_data = sagemaker.session.s3_input(s3_validation_data, distribution='FullyReplicated', 
                             content_type='text/plain', s3_data_type='S3Prefix')
data_channels = {'train': train_data, 'validation': validation_data}
%%local
%%time

bt_model.fit(data_channels, 
             experiment_config={
                      "ExperimentName": sentiment_experiment.experiment_name, 
                      "TrialName": trial.trial_name,
                      "TrialComponentDisplayName": "BlazingText-Training",
                  },
             logs=False)<jupyter_output><empty_output><jupyter_text>Deploy the model and get predictions<jupyter_code>%%local 
text_classifier = bt_model.deploy(initial_instance_count = 1, instance_type = instance_type_smendpoint)
%%local 
review = ["please give this one a miss br br <NAME> and the rest of the cast"
          "rendered terrible performances the show is flat flat flat br br"
          "i don't know how <NAME> could have allowed this one on his plate"
          "he almost seemed to know this wasn't going to work out"
          "and his performance was quite lacklustre so all you madison fans give this a miss"]
tokenized_review = [' '.join(t.split(" ")) for t in review]
#For retrieving the top k predictions, you can set k in the configuration
payload = {"instances" : tokenized_review}
bt_endpoint_name = text_classifier.endpoint
response = sm_runtime.invoke_endpoint(EndpointName=bt_endpoint_name,
                                      ContentType = 'application/json',
                                      Body=json.dumps(payload))
output = json.loads(response['Body'].read().decode('utf-8'))
#make the output readable 
import copy
predictions = copy.deepcopy(output) 
for output in predictions:
    output['label'] = output['label'][0][9:].upper() 
print(predictions)<jupyter_output><empty_output><jupyter_text>Clean up<jupyter_code>%%local
#Clean up resources created as part of this notebook
#delete endpoint
bt_model.delete_endpoint()
# empty s3 bucket we created
s3_bucket_to_remove = "s3://{}".format(bucket)
!aws s3 rm {s3_bucket_to_remove} --recursive<jupyter_output><empty_output><jupyter_text>Delete all sessions created by the notebook<jupyter_code>%%cleanup -f<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0883572189979865, 'fraction_numerical': 0.0074025820206087885, 'mean_word_length': 3.7988064791133844, 'pattern_counts': {'":': 6, '<': 86, '<?xml version=': 0, '>': 87, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '37738', 'license': 'MIT', 'n_tokens_mistral': 4859, 'n_tokens_neox': 4639, 'n_words': 1422, 'path': 'notebooks/Spark EMR-SageMaker/smstudio-pyspark-hive-sentiment-analysis.ipynb', 'repo_name': 'bobbruno/sagemaker-hands-on', 'size': '16892'}