Use Spark and Streaming to predict sentiment.ipynb

{
    "nbformat_minor": 1, 
    "cells": [
        {
            "source": "<div style=\"background:#F5F7FA; height:100px; padding: 2em; font-size:14px;\">\n<span style=\"font-size:18px;color:#152935;\">Want to do more?</span><span style=\"border: 1px solid #3d70b2;padding: 15px;float:right;margin-right:40px; color:#3d70b2; \"><a href=\"https://ibm.co/wsnotebooks\" target=\"_blank\" style=\"color: #3d70b2;text-decoration: none;\">Sign Up</a></span><br>\n<span style=\"color:#5A6872;\"> Try out this notebook with your free trial of IBM Watson Studio.</span>\n</div>", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<table style=\"border: none\" align=\"left\">\n   <tr style=\"border: none\">\n      <th style=\"border: none\"><font face=\"verdana\" size=\"5\" color=\"black\"><b>Use Spark and Streaming to predict sentiment</b></th>\n      <th style=\"border: none\"><img src=\"https://github.com/pmservice/customer-satisfaction-prediction/blob/master/app/static/images/ml_icon_gray.png?raw=true\" alt=\"Watson Machine Learning icon\" height=\"40\" width=\"40\"></th>\n   <tr style=\"border: none\">\n       <th style=\"border: none\"><img src=\"https://github.com/pmservice/wml-sample-models/blob/master/spark/sentiment-prediction/images/sentiment_banner_3-04.png?raw=true\" width=\"600\" alt=\"Icon\"> </th>\n   </tr>\n</table>", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "This notebook contains steps and code to develop a predictive model, and start scoring new data. This notebook introduces commands for getting data and for basic data cleaning and exploration, pipeline creation, model training, model persistance to Watson Machine Learning repository, model deployment, and scoring.\n\nSome familiarity with Python is helpful. This notebook uses Python 3.5 and Apache\u00ae Spark 2.\n\nYou will use a data set named **Twitter Sentiment**, which includes many tweets and their binary labels (e.g. POSITIVE or NEGATIVE). Use the details of this data set to predict sentiment of tweets.\n\n## Learning goals\n\nThe learning goals of this notebook are:\n\n-  Load a CSV file into an Apache\u00ae Spark DataFrame.\n-  Explore data.\n-  Prepare data for training and evaluation.\n-  Create an Apache\u00ae Spark machine learning pipeline.\n-  Train and evaluate a model.\n-  Persist a pipeline and model in Watson Machine Learning repository.\n-  Explore and visualize prediction result using the plotly package.\n-  Deploy a model for streaming using Wastson Machine Learning API.\n-  Working with IBM Cloud MessageHub.\n\n\n## Contents\n\nThis notebook contains the following parts:\n\n1.\t[Setup](#setup)\n2.\t[Load and explore data](#load)\n3.\t[Create spark ml model](#model)\n4.\t[Persist model](#persistence)\n5.\t[Predict locally and visualize](#visualization)\n6.\t[Deploy and score in a Cloud](#scoring)\n7.\t[Summary and next steps](#summary)", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<a id=\"setup\"></a>\n## 1. Set up the environment\n\nBefore you use the sample code in this notebook, you must perform the following setup tasks:\n\n### Required IBM Cloud services\n\n-  Create a [Watson Machine Learning (WML) Service](https://console.ng.bluemix.net/catalog/services/ibm-watson-machine-learning/) instance (a free plan is offered and information about how to create the instance is [here](https://dataplatform.ibm.com/docs/content/analyze-data/wml-setup.html)). \n-  Create a [MessageHub Service](https://console.bluemix.net/catalog) instance (free plan is offered).\n-  Create two topics in MessageHub: **streamingi** and **streamingo**. \n-  Make sure that you are using a supported Spark 2.x kernel.\n\n**Note**:  Get your credentials from Manage Credentials section of Watson Machine Learning and MessageHub Service.\n\n### Twitter Sentiment Data\n\nThe Twitter Sentiment data is available on [project's Github page](https://github.com/pmservice/wml-sample-models/tree/master/spark/sentiment-prediction).", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<a id=\"load\"></a>\n## 2. Load and explore data", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this section you will load the data as an Apache\u00ae Spark DataFrame and perform a basic exploration.\n\n**Tip**: First, you need to install required packages. You can do it by running the following code. You can run the code below after changing **cell type** to `code`.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "!pip install wget --user ", 
            "cell_type": "raw", 
            "metadata": {}
        }, 
        {
            "execution_count": 1, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "import wget\nfrom pprint import PrettyPrinter\n\npp = PrettyPrinter(indent=2, depth=3).pprint"
        }, 
        {
            "execution_count": 2, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "trainingTweets.csv\n"
                }
            ], 
            "source": "link_to_data = 'https://github.com/pmservice/wml-sample-models/raw/master/spark/sentiment-prediction/data/trainingTweets.csv'\nfilename = wget.download(link_to_data)\n\nprint(filename)"
        }, 
        {
            "source": "The csv file trainingTweets.csv is availble on gpfs now. Load the file to Apache\u00ae Spark DataFrame using code below.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 3, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "spark = SparkSession.builder.getOrCreate()\n\ndf_data = spark.read\\\n  .format('org.apache.spark.sql.execution.datasources.csv.CSVFileFormat')\\\n  .option('header', 'true')\\\n  .option('inferSchema', 'true')\\\n  .load(filename)"
        }, 
        {
            "source": "Explore the loaded data by using Apache\u00ae Spark DataFrame methods:\n-  print schema\n-  print first ten records\n-  count all records", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 4, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "root\n |-- id: integer (nullable = true)\n |-- text: string (nullable = true)\n |-- label: integer (nullable = true)\n\n"
                }
            ], 
            "source": "df_data.printSchema()"
        }, 
        {
            "source": "As you can see, the data contains 3 fields. ``label`` field is the one we would like to classify tweets.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 5, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "+---+--------------------+-----+\n| id|                text|label|\n+---+--------------------+-----+\n|  1|realdonaldtrump s...|    1|\n|  2|cnnvideo hillaryc...|    1|\n|  3|realdonaldtrump t...|    1|\n|  4|sensanders the am...|    1|\n|  5|billburton realdo...|    0|\n|  6|reince hillarycli...|    0|\n|  7|bentechpro realdo...|    1|\n|  8|dahbigj hopeobama...|    0|\n|  9|theosmelek thuddy...|    0|\n| 10|realdonaldtrump r...|    0|\n+---+--------------------+-----+\nonly showing top 10 rows\n\n"
                }
            ], 
            "source": "df_data.show(n=10)"
        }, 
        {
            "execution_count": 6, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "Total number of records: 5987\n"
                }
            ], 
            "source": "print(\"Total number of records: {count}\".format(count=df_data.count()))"
        }, 
        {
            "source": "Data set contains 5987 records.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<a id=\"model\"></a>\n## 3. Create an Apache\u00ae Spark machine learning model\n\nIn this section you will learn how to prepare data, create an Apache\u00ae Spark machine learning pipeline, and train a model.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "### 3.1: Prepare data\n\nIn this subsection you will split your data into: \n-  The train data set, which is the largest group, is used for training.\n-  The test data set will be used for model evaluation and is used to test the assumptions of the model.\n-  The predict data set will be used for prediction.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 7, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "Number of training records: 4783\nNumber of testing records: 1076\nNumber of prediction records: 128\n"
                }
            ], 
            "source": "splitted_data = df_data.randomSplit([0.8, 0.18, 0.02], 24)\ntrain_data = splitted_data[0]\ntest_data = splitted_data[1]\npredict_data = splitted_data[2]\n\nprint(\"Number of training records: {count}\".format(count=train_data.count()))\nprint(\"Number of testing records: {count}\".format(count=test_data.count()))\nprint(\"Number of prediction records: {count}\".format(count=predict_data.count()))"
        }, 
        {
            "source": "### 3.2: Create pipeline and train a model", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this section you will create an Apache\u00ae Spark machine learning pipeline and then train the model.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In the first step you need to import the Apache\u00ae Spark machine learning packages that will be needed in the subsequent steps.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 8, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "from pyspark.ml.feature import Tokenizer, OneHotEncoder, StringIndexer, HashingTF\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.evaluation import BinaryClassificationEvaluator\nfrom pyspark.ml.tuning import ParamGridBuilder, CrossValidator\nfrom pyspark.ml import Pipeline, Model"
        }, 
        {
            "source": "In the data preprocessing step, convert all the string fields to numeric ones by using **Tokenizer** and then **HashingTF** transformer.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 9, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "tokenizer = Tokenizer(inputCol=\"text\", outputCol=\"words\")\nhashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol=\"features\")"
        }, 
        {
            "source": "Next, define estimator you want to use for classification. Logistic Regression is used in the following example.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 10, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "lr = LogisticRegression(maxIter=10, regParam=0.01)"
        }, 
        {
            "source": "Let's build the pipeline now. A pipeline consists of transformers and an estimator.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 11, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "pipeline_lr = Pipeline(stages=[tokenizer, hashingTF, lr])"
        }, 
        {
            "source": "Now, you can train your Logistic Regression model by using previously defined **pipeline** and **train data**.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 12, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "model_lr = pipeline_lr.fit(train_data)"
        }, 
        {
            "source": "You can evaluate the model on the test data. Area under ROC will be used as evaluation metrics.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 13, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "Accuracy = 0.762\n"
                }
            ], 
            "source": "predictions = model_lr.transform(test_data)\nevaluatorRF = BinaryClassificationEvaluator(labelCol=\"label\", rawPredictionCol=\"prediction\", metricName=\"areaUnderROC\")\naccuracy = evaluatorRF.evaluate(predictions)\n\nprint(\"Accuracy = {acc:4.3f}\".format(acc=accuracy))"
        }, 
        {
            "source": "Now you can tune your model to achieve better accuracy. For simplicity of this notebook, tuning section is omitted.", 
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }
        }, 
        {
            "source": "<a id=\"persistence\"></a>\n## 4. Persist model", 
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }
        }, 
        {
            "source": "In this section you will learn how to store your pipeline and model in Watson Machine Learning repository by using python client libraries.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "First, you must install and import Watson Machine Learning client libraries.", 
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }
        }, 
        {
            "execution_count": 14, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "!rm -rf $PIP_BUILD/watson-machine-learning-client"
        }, 
        {
            "execution_count": 15, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "Requirement already up-to-date: watson-machine-learning-client in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages\nRequirement already up-to-date: tqdm in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: tabulate in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: urllib3 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: certifi in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: pandas in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: lomond in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: requests in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from watson-machine-learning-client)\nRequirement already up-to-date: python-dateutil>=2 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from pandas->watson-machine-learning-client)\nRequirement already up-to-date: numpy>=1.9.0 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from pandas->watson-machine-learning-client)\nRequirement already up-to-date: pytz>=2011k in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from pandas->watson-machine-learning-client)\nRequirement already up-to-date: six>=1.10.0 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from lomond->watson-machine-learning-client)\nRequirement already up-to-date: chardet<3.1.0,>=3.0.2 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from requests->watson-machine-learning-client)\nRequirement already up-to-date: idna<2.7,>=2.5 in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s86b-18e61b28c674e4-3fbaf243aed6/.local/lib/python3.5/site-packages (from requests->watson-machine-learning-client)\n"
                }
            ], 
            "source": "!pip install watson-machine-learning-client --upgrade"
        }, 
        {
            "execution_count": 16, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "from watson_machine_learning_client import WatsonMachineLearningAPIClient"
        }, 
        {
            "source": "Authenticate to the Watson Machine Learning service on IBM Cloud.\n\n**Tip**: Authentication information (your credentials) can be found in the [Service Credentials](https://console.bluemix.net/docs/services/service_credentials.html#service_credentials) tab of the service instance that you created on IBM Cloud. \nIf you cannot see the **instance_id** field in **Service Credentials**, click **New credential (+)** to generate new authentication information. \n\n**Action**: Enter your Watson Machine Learning service instance credentials here.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 15, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "wml_credentials={\n  \"url\": \"https://ibm-watson-ml.mybluemix.net\",\n  \"access_key\": \"****\",\n  \"username\": \"****\",\n  \"password\": \"****\",\n  \"instance_id\": \"****\"\n}"
        }, 
        {
            "execution_count": 17, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "# The code was removed by DSX for sharing."
        }, 
        {
            "source": "Create WML client and authorize it.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 18, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "client = WatsonMachineLearningAPIClient(wml_credentials)"
        }, 
        {
            "execution_count": 19, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "execution_count": 19, 
                    "metadata": {}, 
                    "data": {
                        "text/plain": "'1.0.53'"
                    }, 
                    "output_type": "execute_result"
                }
            ], 
            "source": "client.version"
        }, 
        {
            "source": "### 4.1: Save pipeline and model", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this subsection you will learn how to save pipeline and model artifacts to your Watson Machine Learning instance.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 20, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "9faf9502-87ca-4a15-8581-060cc84ccc5b\n"
                }
            ], 
            "source": "saved_model_details = client.repository.store_model(model=model_lr, meta_props={\"name\": \"Sentiment Prediction Model\"}, training_data=train_data, pipeline=pipeline_lr)\n\nmodel_uid = client.repository.get_model_uid(saved_model_details)\nprint(model_uid)"
        }, 
        {
            "source": "Check model details:", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 21, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "{'metadata': {'created_at': '2018-03-29T10:37:16.069Z', 'url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b', 'modified_at': '2018-03-29T10:37:16.143Z', 'guid': '9faf9502-87ca-4a15-8581-060cc84ccc5b'}, 'entity': {'latest_version': {'created_at': '2018-03-29T10:37:16.143Z', 'url': 'https://ibm-watson-ml.mybluemix.net/v3/ml_assets/models/9faf9502-87ca-4a15-8581-060cc84ccc5b/versions/42cec03e-bdde-4996-a0b9-118e143f804b', 'guid': '42cec03e-bdde-4996-a0b9-118e143f804b'}, 'runtime_environment': 'spark-2.1', 'name': 'Sentiment Prediction Model', 'learning_configuration_url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b/learning_configuration', 'model_type': 'mllib-2.1', 'input_data_schema': {'type': 'struct', 'fields': [{'metadata': {}, 'nullable': True, 'type': 'integer', 'name': 'id'}, {'metadata': {}, 'nullable': True, 'type': 'string', 'name': 'text'}]}, 'feedback_url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b/feedback', 'label_col': 'label', 'deployments': {'url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b/deployments', 'count': 0}, 'training_data_schema': {'type': 'struct', 'fields': [{'metadata': {}, 'nullable': True, 'type': 'integer', 'name': 'id'}, {'metadata': {}, 'nullable': True, 'type': 'string', 'name': 'text'}, {'metadata': {}, 'nullable': True, 'type': 'integer', 'name': 'label'}]}, 'learning_iterations_url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b/learning_iterations', 'evaluation_metrics_url': 'https://ibm-watson-ml.mybluemix.net/v3/wml_instances/3f6e5c2b-4880-46aa-9d79-62e90ccc9d56/published_models/9faf9502-87ca-4a15-8581-060cc84ccc5b/evaluation_metrics', 'author': {}}}\n"
                }
            ], 
            "source": "print(saved_model_details)"
        }, 
        {
            "source": "### 4.2: Load model", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this subsection you will learn how to load back saved model from specified instance of Watson Machine Learning.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 22, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "loaded_model = client.repository.load(model_uid)"
        }, 
        {
            "source": "You can check the type of model. As it is the same model you saved, you can use it for local scoring.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 23, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "<class 'pyspark.ml.pipeline.PipelineModel'>\n"
                }
            ], 
            "source": "print(type(loaded_model))"
        }, 
        {
            "source": "As you can see the type is correct. You have already learned how save and load the model from Watson Machine Learning repository.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<a id=\"visualization\"></a>\n## 5. Predict locally and visualize", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this section you will learn how to score test data using loaded model and visualize the prediction results with plotly package.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "### 5.1: Make local prediction using previously loaded model and test data", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this subsection you will score ``predict_data`` data set.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 24, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "predictions = loaded_model.transform(predict_data)"
        }, 
        {
            "source": "Preview the results by calling the *show()* method on the predictions DataFrame.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 25, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "+---+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+\n| id|                text|label|               words|            features|       rawPrediction|         probability|prediction|\n+---+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+\n|254|realdonaldtrump t...|    1|[realdonaldtrump,...|(262144,[4312,961...|[-1.3150612511415...|[0.21164114610918...|       1.0|\n|256|realdonaldtrump j...|    0|[realdonaldtrump,...|(262144,[14,13396...|[5.48207446382384...|[0.99585655006955...|       0.0|\n|296|realdonaldtrump t...|    1|[realdonaldtrump,...|(262144,[15889,21...|[-1.3175186267989...|[0.21123142555360...|       1.0|\n|312|sensanders keep t...|    0|[sensanders, keep...|(262144,[32890,91...|[3.29411163264101...|[0.96422625174639...|       0.0|\n|362|katiedaviscourt i...|    0|[katiedaviscourt,...|(262144,[16332,21...|[3.83815328742004...|[0.97892057960460...|       0.0|\n+---+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+\nonly showing top 5 rows\n\n"
                }
            ], 
            "source": "predictions.show(5)"
        }, 
        {
            "source": "By tabulating a count, you can see the split by sentiment.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 26, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "+-----+-----+\n|label|count|\n+-----+-----+\n|    1|   49|\n|    0|   79|\n+-----+-----+\n\n"
                }
            ], 
            "source": "predictions.select(\"label\").groupBy(\"label\").count().show()"
        }, 
        {
            "source": "### 5.2: Sample visualization of data with Plotly package", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this subsection you will explore prediction results with Plotly, which is an online analytics and data visualization tool.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "**Tip**: First, you need to install required packages. You can do it by running the following code. Run it only one time.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "!pip install plotly --user\n!pip install cufflinks --user", 
            "cell_type": "raw", 
            "metadata": {}
        }, 
        {
            "execution_count": 27, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "display_data", 
                    "data": {
                        "text/html": "<script>requirejs.config({paths: { 'plotly': ['https://cdn.plot.ly/plotly-latest.min']},});if(!window.Plotly) {{require(['plotly'],function(plotly) {window.Plotly=plotly;});}}</script>", 
                        "text/vnd.plotly.v1+html": "<script>requirejs.config({paths: { 'plotly': ['https://cdn.plot.ly/plotly-latest.min']},});if(!window.Plotly) {{require(['plotly'],function(plotly) {window.Plotly=plotly;});}}</script>"
                    }, 
                    "metadata": {}
                }
            ], 
            "source": "import sys\nimport pandas\nimport plotly.plotly as py\nfrom plotly.offline import download_plotlyjs, init_notebook_mode, plot, iplot\nimport cufflinks as cf\nimport plotly.graph_objs as go\n\ninit_notebook_mode(connected=True)\nsys.path.append(\"\".join([os.environ[\"HOME\"]])) "
        }, 
        {
            "source": "You have to convert the Apache Spark DataFrame to a Pandas DataFrame to be used by ploting function.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 28, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "predictions_pdf = predictions.select(\"prediction\", \"label\", \"text\").toPandas()\ncumulative_stats = predictions_pdf.groupby(['label']).count()\nlabels_data_plot = cumulative_stats.index\nvalues_data_plot = cumulative_stats['text']"
        }, 
        {
            "source": "Plot a pie chart that shows the predicted sentiment label.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 29, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "display_data", 
                    "data": {
                        "text/html": "<div id=\"cfad1f7a-4495-44fb-b597-b2043ba0e139\" style=\"height: 525px; width: 100%;\" class=\"plotly-graph-div\"></div><script type=\"text/javascript\">require([\"plotly\"], function(Plotly) { window.PLOTLYENV=window.PLOTLYENV || {};window.PLOTLYENV.BASE_URL=\"https://plot.ly\";Plotly.newPlot(\"cfad1f7a-4495-44fb-b597-b2043ba0e139\", [{\"type\": \"pie\", \"values\": [79, 49], \"labels\": [0, 1]}], {\"title\": \"Sentiment\"}, {\"linkText\": \"Export to plot.ly\", \"showLink\": true})});</script>", 
                        "text/vnd.plotly.v1+html": "<div id=\"cfad1f7a-4495-44fb-b597-b2043ba0e139\" style=\"height: 525px; width: 100%;\" class=\"plotly-graph-div\"></div><script type=\"text/javascript\">require([\"plotly\"], function(Plotly) { window.PLOTLYENV=window.PLOTLYENV || {};window.PLOTLYENV.BASE_URL=\"https://plot.ly\";Plotly.newPlot(\"cfad1f7a-4495-44fb-b597-b2043ba0e139\", [{\"type\": \"pie\", \"values\": [79, 49], \"labels\": [0, 1]}], {\"title\": \"Sentiment\"}, {\"linkText\": \"Export to plot.ly\", \"showLink\": true})});</script>", 
                        "application/vnd.plotly.v1+json": {
                            "data": [
                                {
                                    "type": "pie", 
                                    "labels": [
                                        0, 
                                        1
                                    ], 
                                    "values": [
                                        79, 
                                        49
                                    ]
                                }
                            ], 
                            "layout": {
                                "title": "Sentiment"
                            }
                        }
                    }, 
                    "metadata": {}
                }
            ], 
            "source": "product_data = [go.Pie(\n            labels=labels_data_plot,\n            values=values_data_plot,\n    )]\n\nproduct_layout = go.Layout(\n    title='Sentiment',\n)\n\nfig = go.Figure(data=product_data, layout=product_layout)\niplot(fig)"
        }, 
        {
            "source": "<a id=\"scoring\"></a>\n## 6. Deploy and score in a Cloud", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "In this section you will learn how to create streaming deployment by using the Watson Machine Learning REST API and work with IBM Cloud MessageHub. \nFor more information about REST APIs, see the [Swagger Documentation](http://watson-ml-api.mybluemix.net/).", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "### 6.1: Create streaming deployment", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "#### Get the list of published models", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 30, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "------------------------------------  ----------------------------------  ------------------------  -----------------\nGUID                                  NAME                                CREATED                   FRAMEWORK\n43986a55-0f0e-493b-9b98-0712f918b0ba  MNIST CNN model                     2018-03-15T16:12:06.503Z  tensorflow-1.5\na5a952e1-fa6f-4c64-88f7-293b367b7e1f  MNIST best model                    2018-03-16T10:12:16.773Z  tensorflow-1.5\nc755c740-dab6-4d5d-b873-1060885d55d5  MNIST best model                    2018-03-16T10:12:31.990Z  tensorflow-1.5\n861b6b73-6861-49c0-867c-d171d6d9d545  MNIST best model                    2018-03-16T13:00:40.461Z  tensorflow-1.5\nf7219c8f-fccb-4d26-a1bd-f51544090eae  training-S3ttnlgig_2 best model     2018-03-16T13:32:33.747Z  tensorflow-1.5\n327242a2-d2c0-400f-b673-4f33ad7ec17d  product-line-prediction             2018-03-16T18:08:31.260Z  mllib-2.0\na0cb79ca-e937-4cbf-b7e5-23dad2d2ecec  drug-selection                      2018-03-16T18:10:25.046Z  mllib-2.0\n97822727-4bd1-4df0-95df-8ff9536ea01d  My cool mnist model                 2018-03-28T09:02:31.509Z  tensorflow-1.5\n84070d90-6db7-4a12-8da6-26aaf18e9378  My cool mnist model tf-horovod      2018-03-28T11:15:53.446Z  tensorflow-1.5\nb6e59ce5-7d24-4ce1-90f8-f6a918ad71ca  My cool mnist model tf-distributed  2018-03-28T11:15:56.061Z  tensorflow-1.5\n91149430-5588-4ee6-ac6a-695cd2392aa8  My cool mnist model tf-ddl          2018-03-28T11:15:59.813Z  tensorflow-1.5\n5bcf67ab-cd88-49c9-99fd-99dde60eff70  SPSS model for Churn prediction     2018-03-28T13:11:28.164Z  spss-modeler-17.1\n9fb96bc1-bc12-4717-a03f-9b009a32d578  SPSS model for Churn prediction     2018-03-28T13:20:40.791Z  spss-modeler-18.1\n0da45aa2-5378-4dc1-8617-3d209499f466  my spark model                      2018-03-28T14:49:01.055Z  mllib-2.1\n298832f7-d21d-4a07-a028-b5716dba1264  Best Heart Drug Selection           2018-03-29T08:35:31.303Z  mllib-2.1\n70a8b8ad-ef04-4d9f-a1e7-679719577b62  Product line model                  2018-03-29T09:21:46.826Z  mllib-2.1\n9faf9502-87ca-4a15-8581-060cc84ccc5b  Sentiment Prediction Model          2018-03-29T10:37:16.069Z  mllib-2.1\n------------------------------------  ----------------------------------  ------------------------  -----------------\n"
                }
            ], 
            "source": "client.repository.list_models()"
        }, 
        {
            "source": "#### Create streaming deployment for published model", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "**Action**: Put credentials from your instance of **Spark Service** in cell bellow.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 29, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "spark_credentials={\n  \"tenant_id\": \"****\",\n  \"tenant_id_full\": \"****\",\n  \"cluster_master_url\": \"https://spark.bluemix.net\",\n  \"tenant_secret\": \"****\",\n  \"instance_id\": \"****\",\n  \"plan\": \"ibm.SparkService.PayGoPersonal\"\n}"
        }, 
        {
            "execution_count": 31, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "# The code was removed by DSX for sharing."
        }, 
        {
            "source": "Encode the spark credentials.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 32, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "import base64\nimport json\n\nspark_instance = {\n    \"credentials\": spark_credentials,\n    \"version\": \"2.0\"\n}\nencoded_spark_instance = base64.b64encode(json.dumps(spark_instance).encode('utf-8'))"
        }, 
        {
            "source": "#### Create streaming deployment for published model", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 33, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "input_topic_name = 'streamingi'\noutput_topic_name = 'streamingo'"
        }, 
        {
            "source": "**Action**: Please paste credentials of your **MessageHub Service** to cell bellow.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 32, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "messagehub_credentials={\n  \"instance_id\": \"****\",\n  \"mqlight_lookup_url\": \"https://mqlight-lookup-prod01.messagehub.services.us-south.bluemix.net/Lookup?serviceId=fa4d499e-b908-4982-84f0-69c30c1b30a0\",\n  \"api_key\": \"****\",\n  \"kafka_admin_url\": \"https://kafka-admin-prod01.messagehub.services.us-south.bluemix.net:443\",\n  \"kafka_rest_url\": \"https://kafka-rest-prod01.messagehub.services.us-south.bluemix.net:443\",\n  \"kafka_brokers_sasl\": [\n    \"kafka03-prod01.messagehub.services.us-south.bluemix.net:9093\",\n    \"kafka04-prod01.messagehub.services.us-south.bluemix.net:9093\",\n    \"kafka01-prod01.messagehub.services.us-south.bluemix.net:9093\",\n    \"kafka02-prod01.messagehub.services.us-south.bluemix.net:9093\",\n    \"kafka05-prod01.messagehub.services.us-south.bluemix.net:9093\"\n  ],\n  \"user\": \"****\",\n  \"password\": \"****\"\n}"
        }, 
        {
            "execution_count": 34, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "# The code was removed by DSX for sharing."
        }, 
        {
            "execution_count": null, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "import requests\n\nmessagehub_instance = {x: messagehub_credentials[x] for x in messagehub_credentials.keys() if x != \"instance_id\"}\nheader_streaming = {'Content-Type': 'application/json', 'Authorization': 'Bearer {token}'.format(token=client.wml_token), 'X-Spark-Service-Instance': encoded_spark_instance}\npayload_streaming = {\"type\":\"stream\", \"name\": \"Sentiment Prediction\", \"description\": \"Streaming Deployment\", \n                     \"input\": {\"connection\": messagehub_instance, \"source\": { \"type\": \"kafka\", \"topic\": input_topic_name}}, \n                     \"output\": {\"connection\": messagehub_instance, \"target\": {\"type\": \"kafka\", \"topic\": output_topic_name}}}\n\nendpoint_deployments = client.repository.get_model_url(saved_model_details) + \"/deployments\"\n\nresponse_streaming = requests.post(endpoint_deployments, json=payload_streaming, headers=header_streaming)\n\nprint(response_streaming)\npp(json.loads(response_streaming.text))"
        }, 
        {
            "source": "Now, you have streaming deployment of your model ready to make predictions.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "<a id=\"messagehub\"></a>\n## 7. Making predictions with streaming deployment", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "When we want to make predictions with streaming deployment of our model we have to push messages to input topic. Streaming deployment will process that message and write results to output topic.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "#### Get the list of topics from MessageHub", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 34, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "<Response [200]>\n['strimingi', 'strimingo']\n"
                }
            ], 
            "source": "messagehub_topic_header = {'X-Auth-Token': messagehub_credentials['api_key']}\nmessagehub_endpoint = 'https://kafka-rest-prod01.messagehub.services.us-south.bluemix.net:443/topics/'\n\nmessagehub_response = requests.get(messagehub_endpoint, headers=messagehub_topic_header)\n\nprint(messagehub_response)\npp(json.loads(messagehub_response.text))"
        }, 
        {
            "source": "**Note**: You should see two topics: **streamingi** and **streamingo** created in setup step.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "### Send scoring data to input topic", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "We have to encode input data to send it to stream.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": 36, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [
                {
                    "output_type": "stream", 
                    "name": "stdout", 
                    "text": "b'eyJpZCI6IDEsICJ0ZXh0IjogIk15IGZpcnN0IHN0cmVhbWluZyBkZXBsb3ltZW50ISBJdCBmZWVscyBzbyBhd2Vzb21lISEifQ=='\n"
                }
            ], 
            "source": "encoded_payload_value = base64.b64encode(json.dumps({'id': 1, 'text':'My first streaming deployment! It feels so awesome!!'}).encode('utf-8'))\n\nprint(encoded_payload_value)"
        }, 
        {
            "source": "In order to push encoded data to our topic, we will use below curl call.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": null, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "messagehub_header = dict({'Content-Type': 'application/vnd.kafka.binary.v1+json'}, **messagehub_topic_header)\ninput_payload = {'records':[{ 'key': 'null', 'value': encoded_payload_value}]}\nmessagehub_input_endpoint = messagehub_endpoint + 'streamingi'\n\nresponse_message_input = requests.post(messagehub_input_endpoint, json=input_payload, headers=messagehub_header)\n                                \nprint(response_message_input)                                \npp(json.loads(response_message_input.text))"
        }, 
        {
            "source": "### Receive scored data from output topic", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "Create MessageHub consumer", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": null, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "output_consumer_payload = {\"format\": \"binary\", \"auto.offset.reset\": \"smallest\"}\nmessagehub_output_consumer_endpoint = 'https://kafka-rest-prod01.messagehub.services.us-south.bluemix.net:443/consumers/my_output_consumer'\nmessagehub_output_consumer_response = requests.post(messagehub_output_consumer_endpoint, json=output_consumer_payload, headers=messagehub_header)\n\nprint(messagehub_output_consumer_response)\npp(json.loads(messagehub_output_consumer_response.text))"
        }, 
        {
            "source": "Read messages from output topic", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": null, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "messagehub_read_output_endpoint = \"{uri}/topics/{output_topic}\".format(uri=json.loads(messagehub_output_consumer_response.text).get('base_uri'), output_topic=output_topic_name)\nmessagehub_read_output_response = requests.get(messagehub_read_output_endpoint, headers=messagehub_header)\n\nprint(messagehub_read_output_response)\npp(json.loads(messagehub_read_output_response.text))"
        }, 
        {
            "source": "Decode the output message", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "execution_count": null, 
            "cell_type": "code", 
            "metadata": {}, 
            "outputs": [], 
            "source": "message_output_encoded = json.loads(messagehub_read_output_response.text)[0].get('value')\nmessage_output_decoded = base64.b64decode(message_output_encoded)\n\npp(message_output_decoded)"
        }, 
        {
            "source": "<a id=\"summary\"></a>\n## 8. Summary and next steps     ", 
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }
        }, 
        {
            "source": " You successfully completed this notebook! You learned how to use Apache Spark machine learning as well as Watson Machine Learning for model creation and deployment. Check out our [Online Documentation](https://dataplatform.ibm.com/docs/content/analyze-data/wml-setup.html) for more samples, tutorials, documentation, how-tos, and blog posts. ", 
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }
        }, 
        {
            "source": "### Author\n\n**Umit Mert Cakmak** is a Data Scientist at IBM with a track record of developing enterprise-level applications that substantially increases clients' ability to turn data into actionable knowledge.", 
            "cell_type": "markdown", 
            "metadata": {}
        }, 
        {
            "source": "Copyright \u00a9 2017, 2018 IBM. This notebook and its source code are released under the terms of the MIT License.", 
            "cell_type": "markdown", 
            "metadata": {}
        }
    ], 
    "metadata": {
        "kernelspec": {
            "display_name": "Python 2 with Spark 2.1", 
            "name": "python2-spark21", 
            "language": "python"
        }, 
        "language_info": {
            "mimetype": "text/x-python", 
            "nbconvert_exporter": "python", 
            "version": "2.7.11", 
            "name": "python", 
            "file_extension": ".py", 
            "pygments_lexer": "ipython2", 
            "codemirror_mode": {
                "version": 2, 
                "name": "ipython"
            }
        }
    }, 
    "nbformat": 4
}