Leveraging Unstructured Data with Cloud Dataproc.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# GCP Professional Data Engineer\n",
    "### Leveraging Unstructured Data with Cloud Dataproc on Google Cloud Platform\n",
    "#### Modules:\n",
    "- Introduction to Cloud Dataproc\n",
    "- Running Dataproc Jobs\n",
    "- Leveraging GCP\n",
    "- Analyzing Unstructured Data\n",
    "\n",
    "#### Course Description:\n",
    "\n",
    "This 1-week, accelerated course builds upon previous courses in the Data Engineering on Google Cloud Platform specialization. Through a combination of video lectures, demonstrations, and hands-on labs, you'll learn how to create and manage computing clusters to run Hadoop, Spark, Pig and/or Hive jobs on Google Cloud Platform.  You will also learn how to access various cloud storage options from their compute clusters and integrate Google’s machine learning capabilities into their analytics programs.  \n",
    "\n",
    "In the hands-on labs, you will create and manage Dataproc Clusters using the Web Console and the CLI, and use cluster to run Spark and Pig jobs. You will then create iPython notebooks that integrate with BigQuery and storage and utilize Spark. Finally, you integrate the machine learning APIs into your data analysis.\n",
    "\n",
    "Pre-requisites\n",
    "- Google Cloud Platform Big Data & Machine Learning Fundamentals (or equivalent experience)\n",
    "- Some knowledge of Python"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Module 1: Introduction to Cloud Dataproc\n",
    "### Introducing Cloud Dataproc\n",
    "- Cloud Dataproc is a fast, easy-to-use, fully-managed cloud service for running Apache Spark and Apache Hadoop clusters in a simpler, more cost-efficient way. Operations that used to take hours or days take seconds or minutes instead, and you pay only for the resources you use (with per-second billing). Cloud Dataproc also easily integrates with other Google Cloud Platform (GCP) services, giving you a powerful and complete platform for data processing, analytics and machine learning.\n",
    "\n",
    "### Why Unstructured Data\n",
    "- About 90% of enterprise data that is collected by a business tends to be unstructured. This includes:\n",
    "    - Emails, reviews, text, etc.\n",
    "- Consider the Google Steetview initiative- cars gathering photos at the street level with no immediate impact, yet now is the foundation for one of the most useful datasets available for autonomous cars.\n",
    "    \n",
    "### Why Cloud Dataproc\n",
    "- Horizontal vs vertical scaling\n",
    "- Hadoop origins:\n",
    "    - Based on a whitepaper describing MapReduce, Hadoop is an open-source distributed file system also known as HDFS. Spark is a framework that takes advantage of the distributed file system to effectively process tasks.\n",
    "- Running an onsite Hadoop cluster is costly and often inefficient.\n",
    "- Additional benefits:\n",
    "    - Stateless clusters in <90 seconds\n",
    "    - Supports Hadoop, Spark, Pig, Hive\n",
    "    - High-level APIs for job submission\n",
    "    - Connectors in BigTable,  BigQuery, Cloud Storage\n",
    "\n",
    "### Lab: Create a Dataproc Cluster\n",
    "\n",
    "#### Objectives:\n",
    "- Prepare a bucket for cluster initialization\n",
    "- Create a Dataproc Hadoop Cluster customized to use the Google Cloud API\n",
    "- Enable secure access to the Dataproc cluster\n",
    "- Explore Hadoop operations\n",
    "\n",
    "#### Task 1: Prepare Environment Variables\n",
    "- In the Console, on the Navigation menu () click Compute Engine > VM instances.\n",
    "- Locate the line with the instance called training_vm.\n",
    "- On the far right, under 'connect', Click on SSH to open a terminal window.\n",
    "- In this lab you will enter CLI commands on the training_vm.\n",
    "##### Create the source file for setting and resetting environment variables\n",
    "\n",
    "\n",
    "- In the training_vm SSH terminal window, using your preferred command line editor, create and edit the file to hold your environment variables. For example:\n",
    "- One environment variable that you will set is 'PROJECT_ID' that contains the Google Cloud project ID required to access billable resources.\n",
    "- In the Console, on the Navigation menu () click Home. In the panel with Project Info, the Project ID is listed. You can also find this information in the Qwiklabs tab under Connection Details, where it is labeled GCP Project ID.\n",
    "Add the environment variable to myenv for easy reference.\n",
    "- Dataproc can use a Cloud Storage bucket to stage its files during initialization. You can use this bucket to stage application programs or data for use by Dataproc. The bucket can also host Dataproc initialization scripts and output. The bucket name must be globally unique. Qwiklabs has already created a bucket for you that has the same name as the Project ID, which is already globally unique.\n",
    "- In the Console, on the Navigation menu () click Storage > Browser. Verify that the bucket exists. Notice the default storage class and the location (region) of this bucket. You will be using this region information next.\n",
    "Add the line to myenv to create an environment variable named \"BUCKET\".\n",
    "- You can use BUCKET in CLI commands. And if you need to enter the bucket name <your-bucket> in a text field in Console, you can quickly retrieve the name with echo BUCKET.\n",
    "- You will be creating a Dataproc cluster in a specific region. The Dataproc cluster and the bucket it will use for staging must be in the same region. Since the bucket you are using already exists, you will need to match the environment variable $MYREGION to the bucket region.\n",
    "- You can use find the region used by Qwiklabs on the Qwiklabs tab under Connection Details, labeled QL Region.\n",
    "The zone must be in the same region MYZONE will contain this value.\n",
    "- You can find the zone used by Qwiklabs on the Qwiklabs tab under Connection Details, labeled QL Zone.\n",
    "- Add the environment variables to myenv for easy reference.\n",
    "You will use the browser IP address to enable your local browser to reach the Dataproc cluster.\n",
    "- Find your computer's browser IP address by opening a browser window and viewing http://ip4.me/ Copy the IP address.\n",
    "Add the line to myenv to create an environment variable named BROWSER_IP.\n",
    "- After you have added all three definitions to myenv, and saved the file, use the source command to create the environment variables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-1-959bb4230150>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-1-959bb4230150>\"\u001b[1;36m, line \u001b[1;32m1\u001b[0m\n\u001b[1;33m    cd ~\u001b[0m\n\u001b[1;37m       ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "cd ~ \n",
    "nano myenv\n",
    "PROJECT_ID=<project ID>\n",
    "BUCKET=<project ID>\n",
    "MYREGION=<region>\n",
    "MYZONE=<zone>\n",
    "BROWSER_IP=<your-browser-ip>\n",
    "\n",
    "# set environment variables\n",
    "source myenv\n",
    "# verify variables are set\n",
    "echo $PROJECT_ID\n",
    "echo $MYREGION $MYZONE\n",
    "echo $BUCKET\n",
    "echo $BROWSER_IP"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 2. Create a Dataproc Cluster\n",
    "- In the Console, on the Navigation menu () click Dataproc > Clusters.\n",
    "- Click Create Cluster.\n",
    "- Specify the following, and leave the remaining settings as their defaults:\n",
    "- Click on Preemptible workers, bucket, network, version, initialization, & access options\n",
    "- Specify the following, and leave the remaining settings as their defaults:\n",
    "    - Name: Cluster Dataproc\n",
    "    - Region: <myregion>\n",
    "    - Zone: <myzone>\n",
    "    - Cluster mode: Standard (1 master, n workers)\n",
    "    - (Master node) Machine type: n1-standard-2\n",
    "    - (Master node) Primary disk size: 100GB\n",
    "    - (Worker node) Machine type: n1-standard-1\n",
    "    - (Worker node) Primary disk size: 50GB\n",
    "    - Nodes: 3\n",
    "    - Network tags: hadoop access\n",
    "    - Cloud storage staging bucket: <your bucket>\n",
    "    - Image version: 1.2\n",
    "    - Project access: Allow API access\n",
    "    \n",
    "    \n",
    "- Create.\n",
    "- The cluster will take several minutes to become operational. In the Console, on the Navigation menu () click Dataproc > Clusters.\n",
    "- Click on your cluster, cluster-dataproc. Then click on the VM Instances tab. The instances will become operational before the hadoop software has completed initialization. When a checkmark in a green circle appears next to the name of the cluster, it is operational.\n",
    "\n",
    "#### Task 3: Enable secure access to Dataproc cluster\n",
    "- Create a firewall rule that allows access only to the Master Node from your computer's IP address. Only ports 8088 (Hadoop Job Interface) and 9870 (Hadoop Admin interface) will be permitted.\n",
    "- Port 8042 is the web UI for the node manager on the worker nodes and port 8080 is the default port for Datalab. Datalab is a notebook-based integrated development environment derived from Jupyter notebooks. It is a common tool for developing Dataproc applications. The Serverless Machine Learning on GCP course uses Datalab extensively.\n",
    "- Recall your computer's browser IP address for use in Console.\n",
    "\n",
    "    ```echo $BROWSER_IP```\n",
    "\n",
    "\n",
    "- In the Console, on the Navigation menu () click VPC Network > Firewall rules.\n",
    "- Click Create Firewall Rule.\n",
    "- Specify the following, and leave the remaining settings as their defaults:\n",
    "    - Name: allow-hadoop\n",
    "    - Network: default\n",
    "    - Priority: 1000\n",
    "    - Direction of traffic: Ingress\n",
    "    - Action on match: allow\n",
    "    - Targets: specified target tags\n",
    "    - Target tags: hadoopaccess\n",
    "    - Source IP ranges: <yourIP>32\n",
    "    - Specified ports and protocols tcp:9870;tcp:8088\n",
    "    \n",
    "- Verify that the network tag \"hadoopaccess\" is set on the Master Node. That will apply the firewall rule to the Master Node, giving your laptop access to it.\n",
    "- In the Console, on the Navigation menu () click Compute Engine > VM Instances.\n",
    "- Click on the Master Node, cluster-dataproc-m.\n",
    "- Verify that under Network Tags it lists hadoopaccess.\n",
    "- If the tag is not there, Click EDIT.\n",
    "- Under Network Tags add the tag: hadoopaccess\n",
    "- Click Save.\n",
    "\n",
    "#### Task 4: Explore Hadoop Operations\n",
    "- In the Console, on the Navigation menu () click Compute Engine > VM Instances.\n",
    "- In the list of VM instances, in the row for cluster-dataproc-m, highlight the External IP and copy it.\n",
    "- Open a new browser tab or window and paste the External IP. Add \":8088\" after the IP and press enter. Example: <External IP>:8088 The web page displayed is the Hadoop Applications interface.\n",
    "- Open a new browser tab or window. Paste the External IP. Add \":9870\" after the IP and press return. Example: <External IP>:9870 The webpage displayed is the Hadoop Administration Interface and should look something like this:\n",
    "- Click on the Datanodes tab. This will show you how much capacity is being used on the worker nodes HDFS (Hadoop Distributed File System) and how much capacity remains.\n",
    "- Click on Utilities > Logs. This shows you the Hadoop log files for each node in the cluster. This is where you can go to investigate problems with Hadoop. Use your browser's back button to return to the Hadoop Administration console.\n",
    "- Click on Utilities > Browse the file system. After a few moments the file system will appear in the browser page. You can use this to navigate the files system. In the row that says Owner is hdfs and Group is hadoop, click on the link that says user. -- - Here you can see directories for all the hadoop services.\n",
    "- Leave the JobTracker <External IP>:8088 and the Administration Interface <External IP>:9870 tabs or windows open. You will use them in the next task.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Module 1 Review\n",
    "\n",
    "1.) Which of the following statements is true about Cloud Dataproc?\n",
    "- Lets you run Spark and Hadoop clusters with minimal administrations\n",
    "- Helps you create job-specific clusters w/o HDFS\n",
    "\n",
    "2.) Matching definitions:\n",
    "- Zone: determines the Google data center where compute nodes will be\n",
    "- Preemtible: costs less but may not always be available\n",
    "- Standard cluster mode: Provides 1 master and n workers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Module 2: Running Dataproc Jobs\n",
    "### Running Jobs\n",
    "- Secure Shell (SSH) is a cryptographic network protocol for operating network services securely over an unsecured network.[1] The standard TCP port for SSH is 22. The best known example application is for remote login to computer systems by users.\n",
    "\n",
    "### Lab: Work with structured and semi-structured Data\n",
    "#### Objectives:\n",
    "- Use the Hive CLI and run a Pig job\n",
    "- Hive is used for structured data, similar to SQL\n",
    "- Pig is used for semi-structured data, similar to SQL + scripting\n",
    "\n",
    "#### Task 1: Preparation\n",
    "- A Dataproc cluster has been prepared for you. If you login to GCP before the progress bar reports that the \"Lab is Running\", you may have to wait several minutes for the cluster to transition from \"Provisioning\" to \"Running\" before the cluster completes setup.\n",
    "- You will be performing most of the lab steps from the Master Node of the cluster in an SSH terminal window.\n",
    "- In the Console, on the Navigation menu () click Dataproc > Clusters.\n",
    "- Locate the cluster named dataproc-cluster. Which region and zone is it located in? The region and zone have been selected automatically for you by Qwiklabs.\n",
    "- Notice the Cloud Storage staging bucket defined for this cluster. This bucket has the same name as the project ID, which is a convenient way to make the name globally unique.\n",
    "- Click on the name dataproc-cluster to go to the Cluster details page.\n",
    "- The Cluster details page opens to the \"Overview\" tab. Click on the tab labeled \"VM Instances\".\n",
    "- Open the Master Node terminal\n",
    "- On the line for the VM named dataproc-cluster-m you will see that it has the Role of Master and there is an SSH link next to it. Click on SSH to open a terminal window to the Master Node.\n",
    "\n",
    "#### Task 2. Enable secure web access to the Dataproc cluster\n",
    "- Create a restrictive firewall rule using Target tags, IP address, and protocol\n",
    "- Create a firewall rule that allows access only to the Master Node from your computer's IP address. Only ports 8088 (Hadoop Job Interface) and 9870 (Hadoop Admin interface) will be permitted.\n",
    "- Verify that the network tag is set on the Master Node\n",
    "- Verify that the network tag \"hadoopaccess\" is set on the Master Node. That will apply the firewall rule to the Master Node, giving your laptop access to it.\n",
    "- In the Console, on the Navigation menu () click Compute Engine > VM Instances.\n",
    "- Click on the Master Node, cluster-dataproc-m.\n",
    "- Verify that under Network Tags it lists hadoopaccess.\n",
    "- If the tag is not there, Click EDIT.\n",
    "- Under Network Tags add the tag: hadoopaccess\n",
    "- Click Save.\n",
    "- Identify the browser IP address\n",
    "- You will use the browser IP address to allow your local browser to connect to the Dataproc cluster.\n",
    "- Find your computer's browser IP address by opening a browser window and viewing http://ip4.me/ Copy the IP address.\n",
    "- Create the firewall rule\n",
    "- In the Console, on the Navigation menu () click VPC Network > Firewall rules.\n",
    "- Click Create Firewall Rule.\n",
    "- Specify the following, and leave the remaining settings as their defaults:\n",
    "    - Name: allow-hadoop\n",
    "    - Network: default\n",
    "    - Priority: 1000\n",
    "    - Direction of traffic: Ingress\n",
    "    - Action on match: allow\n",
    "    - Targets: specified target tags\n",
    "    - Target tags: hadoopaccess\n",
    "    - Source IP ranges:  |yourIP|/32\n",
    "    - Specified ports and protocols tcp:9870;tcp:8088\n",
    "- Click create.\n",
    "    \n",
    "#### Task 3. Prepare the data for Hive\n",
    "- Copy sample files to the Master node home directory\n",
    "- The sample files you need are have already been archived on the Master Node. You will need to copy them into your user directory with the following command.\n",
    "- In the Master Node SSH terminal window:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-5-1ca6502de273>, line 2)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-5-1ca6502de273>\"\u001b[1;36m, line \u001b[1;32m2\u001b[0m\n\u001b[1;33m    cp -r /training .\u001b[0m\n\u001b[1;37m                     ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "cd\n",
    "cp -r /training .\n",
    "ls\n",
    "cd training/training-data-analyst/courses/unstructured\n",
    "ls pet*.*\n",
    "# view structured data in text file  \n",
    "cat pet-details.txt\n",
    "# stage data in HDFS\n",
    "hadoop fs -mkdir /pet-details\n",
    "hadoop fs -put pet-details.txt /pet-details"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- In the Console, on the Navigation menu () click Compute Engine > VM Instances.\n",
    "- In the list of VM instances, in the row for cluster-dataproc-m, highlight the External IP and copy it.\n",
    "- Open a new browser tab or window and paste the External IP. Add \":9870\" after the IP and press enter. Example: <External IP>:9870\n",
    "- You should now see the Hadoop Administration interface. Under Utilities, click on Browse the file system. Click on the folder /pet-details.\n",
    "- Notice that the file pet-details.txt is inside /pet-details.\n",
    "- Leave the Hadoop Administration interface open. You will return to it in later steps.\n",
    "    \n",
    "#### Task 4. Explore Hive using the Hive interactive CLI\n",
    "- Use HIVE to access the data in HDFS as if it were in a database\n",
    "- Hive provides a subset of SQL. The way it does this is by maintaining metadata to define a schema on top of the data. This is one way to work with a large amount of distributed data in HDFS using familiar SQL syntax.\n",
    "- In the master node SSH window, make sure you are in the right directory and start the Hive CLI interpreter:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-2-4bc48b4f45bf>, line 2)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-2-4bc48b4f45bf>\"\u001b[1;36m, line \u001b[1;32m2\u001b[0m\n\u001b[1;33m    CREATE DATABASE pets;\u001b[0m\n\u001b[1;37m                  ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "hive # initialize hive CLI interpreter\n",
    "CREATE DATABASE pets;\n",
    "USE pets\n",
    "# create table\n",
    "CREATE TABLE details (Type String, Name String, Breed String, Color String, Weight Int) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';\n",
    "SHOW TABLES;\n",
    "DESCRIBE pets.details;\n",
    "# establish relationship between metadata schema and data in HDFS\n",
    "load data INPATH '/pet-details/pet-details.txt' OVERWRITE INTO TABLE details;\n",
    "# verify that everything is working\n",
    "SELECT * FROM pets.details;\n",
    "# quit HIVE interpreter\n",
    "quit;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Use the Hadoop Administration interface to see how hive works\n",
    "- Hive ingested the pet-details.txt file into a data warehouse format requiring a schema. You will use the Hadoop Administration interface to see this transformation.\n",
    "- Return to the Hadoop Administration interface in the browser.\n",
    "- Under Utilities, click on Browse the file system. Click on the folder /pet-details. The file pet-details.txt is gone.\n",
    "- Under Utilities, click on Browse the file system. Then click on user > hive > warehouse > pets.db > details. The file pet-details.txt has been moved to this location.\n",
    "\n",
    "***Note: Hive is designed for batch jobs and not for transactions. It ingests data into a data warehouse format requiring a schema. It does not support real-time queries, row-level updates, or unstructured data. Some queries may run much slower than others due to the underlying transformations Hive has to implement to simulate SQL.\n",
    "\n",
    "#### Task 5: Run a Pig job\n",
    "\n",
    "- In the master nodes SSH windowm view the Pig application:\n",
    "\n",
    "```cat pet-details.pig```\n",
    "\n",
    "- In line 'x1', the load statement in the application creates a schema on top of the HDFS data file. Lines 'x2' through 'x5' perform transformations on the data. And the last line stores the result in a folder called /GroupedByType in HDFS.\n",
    "- The application expects to find the ingest file in HDFS in the directory /pet-details. Make another copy of the data at that location:\n",
    "\n",
    "```hadoop fs -put pet-details.txt /pet-details```\n",
    "\n",
    "Run the application:\n",
    "\n",
    "```pig < pet-details.pig```\n",
    "\n",
    "- Return to the browser tab containing the Hadoop Applications interface and refresh it, or reopen it with <External-IP>:8088. Notice that Pig generated a Java MapReduce job which is running on the cluster. Click the browser refresh button to watch for job completion.\n",
    "- Return to the browser tab containing the Hadoop Administration interface and refresh it, or reopen it with <External-IP>:9870. Under Utilities, click on Browse the file system. In the resulting list, click on GroupedByType. This is the output directory specified in the Pig application. The file named part-r-00000 is the HDFS file containing the output. You cannot view the contents from here. First, you must download that part to the local file system\n",
    "- Return to the SSH terminal on the Master node, cloud-dataproc-m and make a local output directory and retrieve the results from HDFS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-3-cb82c8aaa442>, line 2)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-3-cb82c8aaa442>\"\u001b[1;36m, line \u001b[1;32m2\u001b[0m\n\u001b[1;33m    mkdir output\u001b[0m\n\u001b[1;37m               ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "cd\n",
    "mkdir output\n",
    "cd output\n",
    "hadoop fs -get /GroupedByType/part* .\n",
    "# view results of the pig job\n",
    "cat part-r-00000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***Note: Pig provides SQL primitives similar to Hive, but in a more flexible scripting language format. Pig can also deal with semi-structured data, such as data having partial schemas, or for which the schema is not yet known. For this reason it is sometimes used for Extract Transform Load (ETL). It generates Java MapReduce jobs. Pig is not designed to deal with unstructured data.\n",
    "\n",
    "### End Lab"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Separation of Storage & Compute\n",
    "### Submitting Jobs\n",
    "### Spark RDDs, Transformations, and Actions\n",
    "### Lab: Working with Spark Jobs\n",
    "### Module 2 Review"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Module 3: Leveraging GCP\n",
    "### Big Query Support\n",
    "### Lab: Leverage GCP\n",
    "#### Objectives:\n",
    "- Explore Spark using PySpark jobs\n",
    "- Using Cloud Storage instead of HDFS\n",
    "- Run a PySpark application from Cloud Storage\n",
    "- Using Python Pandas to add BigQuery to a Spark application\n",
    "\n",
    "#### Task 1: Prepare the Master Node and the Bucket\n",
    "- In the Console, on the Navigation menu () click Dataproc > Clusters.\n",
    "- Locate the cluster named dataproc-cluster.\n",
    "- Click on the name dataproc-cluster to go to the Cluster details page.\n",
    "- The Cluster details page opens to the \"Overview\" tab. Click on the tab labeled \"VM Instances\".\n",
    "- On the line for the VM named dataproc-cluster-m you will see that it has the Role of Master and there is an SSH link next to it. Click on SSH to open a terminal window to the Master Node.\n",
    "- In the Master Node SSH terminal window, type:\n",
    "\n",
    "```\n",
    "cd\n",
    "cp -r /training .\n",
    "ls\n",
    "```\n",
    "\n",
    "#### Note: A Cloud Storage bucket has already been created for you. It has the same name as the Project ID. You will create an environment variable to make it easy to reference the bucket from the command line on the Master Node.\n",
    "\n",
    "- In the console, on the Navigation menu click Storage -> Bucket\n",
    "- In the Master Node SSH terminal window:\n",
    "\n",
    "```\n",
    "BUCKET=<bucket-name>\n",
    "echo $BUCKET\n",
    "```\n",
    "\n",
    "#### Task 2: The two letter lab\n",
    "\n",
    "#### Note: Why would you want to use Cloud Storage instead of HDFS?\n",
    "You can shut down the cluster when you are not running jobs. The storage persists even when the cluster is shut down, so you don't have to pay for the cluster just to maintain data in HDFS.\n",
    "In some cases Cloud Storage provides better performance than HDFS.\n",
    "Cloud Storage does not require the administration overhead of a local file system\n",
    "\n",
    "- Place a copy of your sample data file in a Cloud Storage bucket instead of HDFS.\n",
    "- In the Master Node terminal window, enter the following gsutil command to copy the sample text files to the Cloud Storage bucket.\n",
    "\n",
    "```\n",
    "gsutil cp /training/road-not-taken.txt gs://$BUCKET\n",
    "```\n",
    "\n",
    "- In the SSH terminal for the Master Node, use nano or vi to create the file wordcount.py\n",
    "- Copy and paste the following code into the file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'pyspark'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-f72fd94c7f86>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mpyspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mSparkSession\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0moperator\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0madd\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mre\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Okay Google.\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mModuleNotFoundError\u001b[0m: No module named 'pyspark'"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from operator import add\n",
    "import re\n",
    "\n",
    "print(\"Okay Google.\")\n",
    "\n",
    "spark = SparkSession\\\n",
    "        .builder\\\n",
    "        .appName(\"CountUniqueWords\")\\\n",
    "        .getOrCreate()\n",
    "\n",
    "lines = spark.read.text(\"/sampledata/road-not-taken.txt\").rdd.map(lambda x: x[0])\n",
    "counts = lines.flatMap(lambda x: x.split(' ')) \\\n",
    "                  .filter(lambda x: re.sub('[^a-zA-Z]+', '', x)) \\\n",
    "                  .filter(lambda x: len(x)>1 ) \\\n",
    "                  .map(lambda x: x.upper()) \\\n",
    "                  .map(lambda x: (x, 1)) \\\n",
    "                  .reduceByKey(add) \\\n",
    "                  .sortByKey()\n",
    "output = counts.collect()\n",
    "for (word, count) in output:\n",
    "  print(\"%s = %i\" % (word, count))\n",
    "\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- First, verify that the data file does not exist in HDFS:\n",
    "```\n",
    "hadoop fs -ls\n",
    "```\n",
    "- Next, use the Hadoop file system command to view the files through the hadoop connector to Cloud Storage. This verifies that the connector is working and that the file is available in the bucket.\n",
    "```\n",
    "hadoop fs -ls gs://$BUCKET\n",
    "```\n",
    "- Edit wordcount.py in nano or vi\n",
    "\n",
    "```\n",
    "lines = spark.read.text(\"/sampledata/road-not-taken.txt\").rdd.map(lambda x: x[0])\n",
    "```\n",
    "\n",
    "- With a line the refers to the file in Cloud Storage. Remember to remove \"/sampledata\" because that directory does not exist. Remember to use the actual bucket name and not the environment variable. The Worker Nodes on the cluster where the program will run do not know the value of the local environment variable on the Master Node."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lines = spark.read.text(\"gs://<YOUR-BUCKET>/road-not-taken.txt\").rdd.map(lambda x: x[0])\n",
    "# run the job\n",
    "spark-submit wordcount.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 3: Run a Pyspark application from Cloud Storage\n",
    "\n",
    "- In the previous task you created a PySpark application in a development environment (on the Master Node). You tested the application using spark-submit.\n",
    "- In this task you will migrate the application from the development environment to a production environment. You will stage the working application file in Cloud Storage. And you will run the production job from Console.\n",
    "- In the Master Node terminal, use the following command to copy the tested wordcount.py PySpark application to the bucket.\n",
    "```\n",
    "gsutil cp wordcount.py gs://$BUCKET\n",
    "```\n",
    "\n",
    "- In the Console, on the Navigation menu () click Dataproc > Clusters. Take note of the region where the cluster is located. You will need that in the next steps.\n",
    "- You will also need the bucket name. You can also retrieve the bucket name from the Master Node terminal by entering the following. Highlight the bucket name and copy it.\n",
    "\n",
    "```\n",
    "echo $BUCKET\n",
    "```\n",
    "- In the Console, on the Navigation menu () click Dataproc > Jobs.\n",
    "- Submit job and specifcy the following:\n",
    "    - Region: <your-region>\n",
    "    - Cluster: dataproc-cluster\n",
    "    - Job type: PySpark\n",
    "    - Main python file: gs://<your bucket>/wordcount.py\n",
    "- Submit and end. Check Dataproc -> Jobs for progress.\n",
    "    \n",
    "### End Lab"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Customizing Clusters\n",
    "### Lab: Cluster Automation using CLI Commands\n",
    "#### Objectives:\n",
    "- Create a customized Dataproc cluster using Cloud Shell\n",
    "\n",
    "#### Task 1: Preparation of Env Variables\n",
    "- In the Console, on the Navigation menu () click Compute Engine > VM instances.\n",
    "- Locate the line with the instance called training_vm.\n",
    "- On the far right, under 'connect', Click on SSH to open a terminal window.\n",
    "- In this lab you will enter CLI commands on the training_vm.\n",
    "#### Note: Dataproc can use a Cloud Storage bucket to stage its files during initialization. You can use this bucket to stage application programs or data for use by Dataproc. The bucket can also host Dataproc initialization scripts and output. The bucket name must be globally unique. Qwiklabs has already created a bucket for you that has the same name as the Project ID, which is already globally unique.\n",
    "\n",
    "- In the Console, on the Navigation menu click Storage > Browser. Verify that the bucket exists. Notice the default storage class and the location (region) of this bucket. You will be using this region information next.\n",
    "- On the training_vm SSH terminal, set the BUCKET.\n",
    "```\n",
    "BUCKET=<bucket name>\n",
    "```\n",
    "\n",
    "- You will be creating a Dataproc cluster in a specific region. The Dataproc cluster and the bucket it will use for staging must be in the same region. Since the bucket you are using already exists, you will need to match the environment variable $MYREGION to the bucket region.\n",
    "- You can use find the region used by Qwiklabs on the Qwiklabs tab under Connection Details, labeled QL Region.\n",
    "- The zone must be in the same region $MYZONE will contain this value.\n",
    "- You can find the zone used by Qwiklabs on the Qwiklabs tab under Connection Details, labeled QL Zone.\n",
    "On the training_vm SSH terminal, set the REGION and ZONE..\n",
    "```\n",
    "MYREGION=<region> \n",
    "# example region: us-central1\n",
    "MYZONE=<zone>\n",
    "# example zone: us-central1-a\n",
    "```\n",
    "- One environment variable that you will set is $PROJECT_ID that contains the Google Cloud project ID required to access billable resources.\n",
    "- In the Console, on the Navigation menu () click Home. In the panel with Project Info, the Project ID is listed. You can also find this information in the Qwiklabs tab under Connection Details, where it is labeled GCP Project ID.\n",
    "- On the training_vm SSH terminal, set the PROJECT_ID.\n",
    "```\n",
    "PROJECT_ID=<project ID>\n",
    "```\n",
    "- Find your computer's browser IP address by opening a browser window and viewing http://ip4.me/ Copy the IP address.\n",
    "- Create an environment variable named BROWSER_IP.\n",
    "```\n",
    "BROWSER_IP=<your-browser-ip>\n",
    "```\n",
    "- In the training_vm SSH terminal window.\n",
    "```\n",
    "cd\n",
    "cp -r /training/training-data-analyst .\n",
    "ls\n",
    "```\n",
    "\n",
    "#### Task 2: Customize the Dataproc Initialization Action\n",
    "\n",
    "- Review the cluster customization script.\n",
    "```\n",
    "cd ~/training-data-analyst/courses/unstructured/\n",
    "cat init-script.sh\n",
    "```\n",
    "- Use nano or vi to edit the init-script.sh file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/bin/bash\n",
    "\n",
    "# install Google Python client on all nodes\n",
    "apt-get update\n",
    "apt-get install -y python-pip\n",
    "pip install --upgrade google-api-python-client\n",
    "\n",
    "ROLE=$(/usr/share/google/get_metadata_value attributes/dataproc-role)\n",
    "if [[ \"${ROLE}\" == 'Master' ]]; then\n",
    "   git clone https://github.com/GoogleCloudPlatform/training-data-analyst\n",
    "fi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 3: Create the Dataproc Cluster\n",
    "- Verify that the Cloud Storage bucket exists and that the $BUCKET and $MYZONE environment variables are still set. The bucket will be used by the Dataproc cluster to stage files as the cluster initializes.\n",
    "```\n",
    "echo $BUCKET $MYREGION $MYZONE\n",
    "echo $PROJECT_ID\n",
    "# copy the customization script to the bucket\n",
    "gsutil cp init-script.sh gs://$BUCKET\n",
    "```\n",
    "#### Note: Cloud Storage is a very sophisticated distributed and resilient data service that supports Spark RDDs. It is connected to Dataproc by a petabit bisection bandwidth network enabling the data to be processed from where it is located rather than needing to be copied. So you can use Cloud Storage instead of HDFS.\n",
    "#### Because data in Cloud Storage survives cluster shutdown, if you used it instead of HDFS, you can terminate clusters when they are not being used to reduce the expense. You can schedule the cluster to terminate after it is idle for a period (when the jobs are done).\n",
    "#### https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/scheduled-deletion\n",
    "\n",
    "\n",
    "- In addition to the custom initialization script, you can use initialization scripts that have been predefined. The script located at: gs://dataproc-initialization-actions/datalab/datalab.sh installs Datalab on the Master Node. Datalab is a notebook-based development environment based on Jupyter notebooks.\n",
    "- Notice that this cluster includes two preemptible worker nodes.\n",
    "- Create the custom cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "gcloud dataproc clusters create cluster-custom \\\n",
    "--bucket $BUCKET \\\n",
    "--subnet default \\\n",
    "--zone $MYZONE \\\n",
    "--master-machine-type n1-standard-2 \\\n",
    "--master-boot-disk-size 100 \\\n",
    "--num-workers 2 \\\n",
    "--worker-machine-type n1-standard-1 \\\n",
    "--worker-boot-disk-size 50 \\\n",
    "--num-preemptible-workers 2 \\\n",
    "--image-version 1.2 \\\n",
    "--scopes 'https://www.googleapis.com/auth/cloud-platform' \\\n",
    "--tags customaccess \\\n",
    "--project $PROJECT_ID \\\n",
    "--initialization-actions 'gs://'$BUCKET'/init-script.sh','gs://dataproc-initialization-actions/datalab/datalab.sh'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Options used in this command include security, cost-savings, and flexibility features.\n",
    "\n",
    "--tags: Applies a network tag so you can automate the creation of firewall rules.\n",
    "\n",
    "--scopes: Applies Cloud IAM restrictions and permissions to the cluster.\n",
    "\n",
    "--num-preemptible-workers: Controls the number of low cost worker nodes present.\n",
    "\n",
    "--initialization-actions: Customizes the software on the cluster. \n",
    "\n",
    "#### Options for further study:\n",
    "\n",
    "--no-address, --network, --subnet:\n",
    "\n",
    "VMs only have internal IPs for added security. Requires enabling GCP API private access on the network, establishing specific firewall rules, and passing the subnet.\n",
    "\n",
    "https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/network\n",
    "\n",
    "#### Task 4: Verify Cluster Customization\n",
    "\n",
    "```\n",
    "# verify browser IP address is set as env variable for use in firewall rule\n",
    "echo $BROWSER_IP\n",
    "# create firewall rule\n",
    "gcloud compute \\\n",
    "--project=$PROJECT_ID \\\n",
    "firewall-rules create allow-custom \\\n",
    "--direction=INGRESS \\\n",
    "--priority=1000 \\\n",
    "--network=default \\\n",
    "--action=ALLOW \\\n",
    "--rules=tcp:9870,tcp:8088,tcp:8080 \\\n",
    "--source-ranges=$BROWSER_IP/32 \\\n",
    "--target-tags=customaccess\n",
    "```\n",
    "- Locate the Master Node External IP Address. In the Console, on the Navigation menu () click Dataproc > Clusters. Click on cluster-custom.\n",
    "- Click on VM instances\n",
    "- Click on cluster-custom-m\n",
    "- In the Network Interfaces section, find the External IP. Highlight and copy it.\n",
    "- Open a new browser tab or window. Enter <external IP>:8080 and press return.\n",
    "- You should see the Google Cloud Datalab.\n",
    "- Creating the custom cluster is the objective of this lab. If this was your production environment, your next steps might be:\n",
    "    - Turn the create commands into a script so that you can start up a cluster on demand.\n",
    "    - Add an option to the command to terminate the cluster after a quiet period.\n",
    "    - Turn the firewall rule into a script so that you can enable/disable external (browser) access only when it is required for administration activities.\n",
    "    - Develop and test your application in Datalab notebooks.\n",
    "    - Host the production application in a Cloud Storage bucket and access your data in either Cloud Storage, BigQuery, or Bigtable.\n",
    "    - For capacity, Edit the number of preemptible worker nodes using Console, and the running cluster will adapt.\n",
    "    - Shut down the cluster when not in use, or schedule auto termination.\n",
    "    \n",
    "### End Lab\n",
    "\n",
    "### Module 3 Review\n",
    "\n",
    "1.) Which of the following will you typically NOT use an initialization action script for?\n",
    "\n",
    "- Change the number of workers in the cluster\n",
    "\n",
    "#### Initialization scripts ARE used for:\n",
    "\n",
    "- Copying custom configuration files\n",
    "- Installing software libraries on the master and worker nodes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Module 4: Analyzing Unstructured Data\n",
    "### Infuse Your Business With Machine Learning\n",
    "### Lab: Add Machine Learning\n",
    "#### Objectives:\n",
    "- Add Machine Learning (ML) to a Spark application\n",
    "\n",
    "#### Task 1: Preparation of Dataproc clusters\n",
    "- In the Console, on the Navigation menu () click Dataproc > Clusters.\n",
    "- Locate the cluster named dataproc-cluster. Which region and zone is it located in? The region and zone have been selected automatically for you by Qwiklabs.\n",
    "- Notice the Cloud Storage staging bucket defined for this cluster. This bucket has the same name as the project ID, which is a convenient way to make the name globally unique.\n",
    "- Click on the name dataproc-cluster to go to the Cluster details page.\n",
    "- The Cluster details page opens to the \"Overview\" tab. Click on the tab labeled \"VM Instances\".\n",
    "- On the line for the VM named dataproc-cluster-m you will see that it has the Role of Master and there is an SSH link next to it. Click on SSH to open a terminal window to the Master Node.\n",
    "#### Prepare API Key\n",
    "\n",
    "\n",
    "- In the Console, on the Navigation menu () click APIs & Services > Credentials.\n",
    "- Click on Create Credentials and select API Key\n",
    "- Copy the API Key. In the terminal, create an environment variable for easy recall of the key.\n",
    "\n",
    "\n",
    "```\n",
    "APIKEY=<your-api-key>\n",
    "```\n",
    "#### Dataproc can use a Cloud Storage bucket to stage its files during initialization. You can use this bucket to stage application programs or data for use by Dataproc. The bucket can also host Dataproc initialization scripts and output. The bucket name must be globally unique. Qwiklabs has already created a bucket for you that has the same name as the Project ID, which is already globally unique.\n",
    "\n",
    "- In the Console, on the Navigation menu () click Storage > Browser. Verify that the bucket exists. Notice the default storage class and the location (region) of this bucket. You will be using this region information next.\n",
    "- On the training_vm SSH terminal, set the BUCKET.\n",
    "```\n",
    "BUCKET=<bucket name>\n",
    "```\n",
    "- One environment variable that you will set is $DEVSHELL_PROJECT_ID that contains the Google Cloud project ID required to access billable resources.\n",
    "- In the Console, on the Navigation menu () click Home. In the panel with Project Info, the Project ID is listed. You can also find this information in the Qwiklabs tab under Connection Details, where it is labeled GCP Project ID.\n",
    "On the training_vm SSH terminal, set the DEVSHELL_PROJECT_ID.\n",
    "```\n",
    "DEVSHELL_PROJECT_ID=<project ID>\n",
    "```\n",
    "- Verify that you have these environment variables are set. Do not proceed until they are set:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "echo $DEVSHELL_PROJECT_ID, $BUCKET, $APIKEY\n",
    "export DEVSHELL_PROJECT_ID\n",
    "export BUCKET\n",
    "export APIKEY\n",
    "\n",
    "# Copy application files to training_vm home directory\n",
    "\n",
    "cd\n",
    "cp -r /training/training-data-analyst .\n",
    "ls\n",
    "cd ~/training-data-analyst/courses/unstructured/\n",
    "# run staging script\n",
    "./stagelabs.sh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/bin/bash\n",
    "# Go to the standard location\n",
    "cd ~/training-data-analyst/courses/unstructured/\n",
    "# \"If at first you don't succeed, try, try again.\"\n",
    "#   If this is our first time here, backup the program files\n",
    "#   If this is a subsequent run, restore fresh from backup before proceeding\n",
    "#\n",
    "if [ -d \"backup\" ]; then\n",
    "  cp backup/*dataproc* .\n",
    "else\n",
    "  mkdir backup\n",
    "  cp *dataproc* backup\n",
    "fi\n",
    "# Verify that the environment variables exist\n",
    "#\n",
    "OKFLAG=1\n",
    "if [[ -v $BUCKET ]]; then\n",
    "  echo \"BUCKET environment variable not found\"\n",
    "  OKFLAG=0\n",
    "fi\n",
    "if [[ -v $DEVSHELL_PROJECT_ID ]]; then\n",
    "  echo \"DEVSHELL_PROJECT_ID environment variable not found\"\n",
    "  OKFLAG=0\n",
    "fi\n",
    "if [[ -v $APIKEY ]]; then\n",
    "  echo \"APIKEY environment variable not found\"\n",
    "  OKFLAG=0\n",
    "fi\n",
    "if [ OKFLAG==1 ]; then\n",
    "  # Edit the script files\n",
    "  sed -i \"s/your-api-key/$APIKEY/\" *dataprocML.py\n",
    "  sed -i \"s/your-project-id/$DEVSHELL_PROJECT_ID/\" *dataprocML.py\n",
    "  sed -i \"s/your-bucket/$BUCKET/\" *dataprocML.py\n",
    "  # Copy python scripts to the bucket\n",
    "  gsutil cp *dataprocML.py gs://$BUCKET/\n",
    "  # Copy data to the bucket\n",
    "  gsutil cp gs:\\/\\/cloud-training\\/gcpdei\\/road* gs:\\/\\/$BUCKET\\/sampledata\\/ \n",
    "  gsutil cp gs:\\/\\/cloud-training\\/gcpdei\\/time* gs:\\/\\/$BUCKET\\/sampledata\\/\n",
    "        \n",
    "fi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " This is what the staging script is doing:\n",
    "    - Edits the three python scripts 01_dataprocML.py, 02_dataprocML.py, 03_dataprocML.py, and replaces the APIKEY, BUCKET, and DEVSHELL_PROJECT_ID with the values from the exported environment variables.\n",
    "    - Copies the updated files to your bucket in Cloud Storage, so that Dataproc can access them.\n",
    "    - Copies sample data files to your bucket.\n",
    "    - Verify that the PySpark application files and sample data files are in the bucket.\n",
    "- In the Console, on the Navigation menu () click Storage > Browser.\n",
    "\n",
    "\n",
    "#### Task 2: Natural Language Processing\n",
    "\n",
    "#### The three programs are \"snapshots\" from a development process. Each program builds on and enhances the one before it. Examining and running each program shows you how to progressively develop a Dataproc/Spark + Machine Learning application.\n",
    "#### The sample data is unstructured data. That is, it either lacks structure, or it has a structure that is not suited to the intended purpose. In this lab you will use Machine Learning to identify and associate the data with values, giving it structure and making the data useful.\n",
    "\n",
    "```\n",
    "cd ~/training-data-analyst/courses/unstructured/\n",
    "```\n",
    "\n",
    "- Examine 01-dataprocML.py using editor such as nano. Don't make any changes to the file.\n",
    "- This program is just a Python program. It will run on Dataproc, but it does not make use of any of the big data features. The program creates a sample line of text in memory and then passes it to the Natural Language Processing service for Sentiment Analysis.\n",
    "- The function SentimentAnalysis() is a wrapper around the REST API. This code creates the structured format of the request and passes the request along with the API Key.\n",
    "- Why is the output printed using a json.dumps?\n",
    "- You could do post-processing of the returned data using Python.\n",
    "- The stagelabs.sh script you ran in Task 1 should have replaced the DEVSHELL_PROJECT_ID, BUCKET, and APIKEY with your information from the environment variables.\n",
    "- Run the application\n",
    "- In the Console, on Navigation menu () click Dataproc > Jobs. The click SUBMIT JOB.\n",
    "- You will need to select the region where your cluster is located, and the cluster, dataproc-cluster. The Job Type is PySpark.\n",
    "- In the field for Main python file, enter the path to the application file, which is something like this: gs://<bucket name>/01-dataprocML.py, where you replace <bucket name> with your bucket name.\n",
    "- Click Submit. View the output.\n",
    "    \n",
    "#### Task 3: Load Sample Data\n",
    "\n",
    "- In the terminal, enter the following commands to copy sample files to the Cloud Storage bucket.\n",
    "\n",
    "```\n",
    "gsutil cp /training/road-not-taken.txt gs://$BUCKET/sampledata/road-not-taken.txt\n",
    "```\n",
    "- In the Console, on the Navigation menu () click Storage > Browser.\n",
    "- Click on your bucket.\n",
    "- Click on sampledata\n",
    "- Some files have already been staged.\n",
    "\n",
    "#### Task 4: Testing Sentiment Analysis with Spark\n",
    "\n",
    "- Examine 02-dataprocML.py using editor such as nano. Don't make any changes to the file.\n",
    "- This program uses Spark RDDs. It reads a small sample file and passes it to the Natural Language Processing service for Sentiment Analysis.\n",
    "- Post-processing of the returned data is done in the pipeline using transformations.\n",
    "- In the Console, on Navigation menu () click Dataproc > Jobs. The click SUBMIT JOB.\n",
    " You will need to select the region where your cluster is located, and the cluster, dataproc-cluster. The Job Type is PySpark.\n",
    "- In the field for Main python file, enter the path to the application file, which is something like this: gs://<bucket name>/02-dataprocML.py, where you replace <bucket name> with your bucket name.\n",
    "- Click Submit. View the output.\n",
    "\n",
    "#### Task 5: Doing Something Useful\n",
    "\n",
    "- Examine 03-dataprocML.py using editor such as nano. Don't make any changes to the file.\n",
    "\n",
    "- This program builds on the previous one. Instead of reading a poem it is going to read an entire book. However, it could just as easily read and process an entire library.\n",
    "- Adds filter (in the pipeline) and sort (Python).\n",
    "- This gives a list of the lines in the book with the strongest sentiment, both positive and negative.\n",
    "- Now this was just a book. Imagine how you could use this to sort through social media commentary. For example, consider the feedback left by customers on a shopping website. You could use this kind of data analysis to identify the most admired and most despised products.\n",
    "- Run the application\n",
    "- In the Console, on Navigation menu () click Dataproc > Jobs. The click SUBMIT JOB.\n",
    "- You will need to select the region where your cluster is located, and the cluster, dataproc-cluster. The Job Type is PySpark.\n",
    "- In the field for Main python file, enter the path to the application file, which is something like this: gs://<bucket name>/03-dataprocML.py, where you replace <bucket name> with your bucket name.\n",
    "\n",
    "### End Lab\n",
    "\n",
    "### Module 4 Review\n",
    "\n",
    "1.) Which (one) of these is NOT a good use case for a ML API?\n",
    "- Identify images where your product is shown upside down (requires domain knowledge & orientation)\n",
    "\n",
    "ML API is GOOD for: \n",
    "- Identifying objects in images\n",
    "- Translation of text\n",
    "- Transcribing audio to text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "Missing parentheses in call to 'print'. Did you mean print(APIKEY)? (<ipython-input-3-18443e1df7e1>, line 31)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-3-18443e1df7e1>\"\u001b[1;36m, line \u001b[1;32m31\u001b[0m\n\u001b[1;33m    print APIKEY\u001b[0m\n\u001b[1;37m               ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m Missing parentheses in call to 'print'. Did you mean print(APIKEY)?\n"
     ]
    }
   ],
   "source": [
    "#!/usr/bin/env python\n",
    "#! 01-dataprocML.py\n",
    "# Copyright 2018 Google Inc.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "'''\n",
    "  This program takes a sample text line of text and passes to a Natural Language Processing\n",
    "  services, sentiment analysis, and processes the results in Python.\n",
    "'''\n",
    "import logging\n",
    "import argparse\n",
    "import json\n",
    "import os\n",
    "from googleapiclient.discovery import build\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(\"local\", \"Simple App\")\n",
    "'''\n",
    "You must set these values for the job to run.\n",
    "'''\n",
    "APIKEY=\"removed_API_key\"   # CHANGE\n",
    "print APIKEY\n",
    "PROJECT_ID=\"qwiklabs-gcp-40459444aeb2e780\"  # CHANGE\n",
    "print PROJECT_ID\n",
    "BUCKET=\"qwiklabs-gcp-40459444aeb2e780\"   # CHANGE\n",
    "## Wrappers around the NLP REST interface\n",
    "def SentimentAnalysis(text):\n",
    "    from googleapiclient.discovery import build\n",
    "    lservice = build('language', 'v1beta1', developerKey=APIKEY)\n",
    "    response = lservice.documents().analyzeSentiment(\n",
    "        body={\n",
    "            'document': {\n",
    "                'type': 'PLAIN_TEXT',\n",
    "                'content': text\n",
    "            }\n",
    "        }).execute()\n",
    "    return response\n",
    "## main\n",
    "sampleline = u'There are places I remember, all my life though some have changed.'\n",
    "#\n",
    "# Calling the Natural Language Processing REST interface\n",
    "#\n",
    "results = SentimentAnalysis(sampleline)\n",
    "# \n",
    "#  What is the service returning?\n",
    "\n",
    "    response = lservice.documents().analyzeSentiment(\n",
    "        body={\n",
    "            'document': {\n",
    "                'type': 'PLAIN_TEXT',\n",
    "                'content': text\n",
    "            }\n",
    "        }).execute()\n",
    "    return response\n",
    "## main\n",
    "sampleline = u'There are places I remember, all my life though some have changed.'\n",
    "#\n",
    "# Calling the Natural Language Processing REST interface\n",
    "#\n",
    "results = SentimentAnalysis(sampleline)\n",
    "# \n",
    "#  What is the service returning?\n",
    "#\n",
    "print \"Function returns: \", type(results)\n",
    "print json.dumps(results, sort_keys=True, indent=4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example Dataproc ML Script [2]\n",
    "\n",
    "This program uses Spark RDDs. \n",
    "\n",
    "It reads a small sample file and passes it to the Natural Language Processing service for Sentiment Analysis.\n",
    "Post-processing of the returned data is done in the pipeline using transformations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/usr/bin/env python\n",
    "#! 02-dataprocML.py\n",
    "# Copyright 2018 Google Inc.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "'''\n",
    "  This program reads a text file and passes to a Natural Language Processing\n",
    "  service, sentiment analysis, and processes the results in Spark.\n",
    "'''\n",
    "import logging\n",
    "import argparse\n",
    "import json\n",
    "import os\n",
    "from googleapiclient.discovery import build\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(\"local\", \"Simple App\")\n",
    "'''\n",
    "You must set these values for the job to run.\n",
    "'''\n",
    "APIKEY=\"removed_API_key\"   # CHANGE\n",
    "print APIKEY\n",
    "PROJECT_ID=\"qwiklabs-gcp-40459444aeb2e780\"  # CHANGE\n",
    "print PROJECT_ID\n",
    "BUCKET=\"qwiklabs-gcp-40459444aeb2e780\"   # CHANGE\n",
    "## Wrappers around the NLP REST interface\n",
    "def SentimentAnalysis(text):\n",
    "    from googleapiclient.discovery import build\n",
    "    lservice = build('language', 'v1beta1', developerKey=APIKEY)\n",
    "    response = lservice.documents().analyzeSentiment(\n",
    "        body={\n",
    "            'document': {\n",
    "                'type': 'PLAIN_TEXT',\n",
    "                'content': text\n",
    "            }\n",
    "        }).execute()\n",
    "    return response\n",
    "## main\n",
    "# We could use sc.textFiles(...)\n",
    "#\n",
    "#   However, that will read each line of text as a separate object.\n",
    "#   And using the REST API to NLP for each line will rapidly exhaust the rate-limit quota \n",
    "#   producing HTTP 429 errors\n",
    "#\n",
    "#   Instead, it is more efficient to pass an entire document to NLP in a single call.\n",
    "#\n",
    "#   So we are using sc.wholeTextFiles(...)  \n",
    "\n",
    "\n",
    "    response = lservice.documents().analyzeSentiment(\n",
    "        body={\n",
    "            'document': {\n",
    "                'type': 'PLAIN_TEXT',\n",
    "                'content': text\n",
    "            }\n",
    "        }).execute()\n",
    "    return response\n",
    "## main\n",
    "# We could use sc.textFiles(...)\n",
    "#\n",
    "#   However, that will read each line of text as a separate object.\n",
    "#   And using the REST API to NLP for each line will rapidly exhaust the rate-limit quota \n",
    "#   producing HTTP 429 errors\n",
    "#\n",
    "#   Instead, it is more efficient to pass an entire document to NLP in a single call.\n",
    "#\n",
    "#   So we are using sc.wholeTextFiles(...)\n",
    "#\n",
    "#      This provides a file as a tuple.\n",
    "#      The first element is the file pathname, and second element is the content of the file.\n",
    "#\n",
    "sample = sc.wholeTextFiles(\"gs://{0}/sampledata/road-not-taken.txt\".format(BUCKET))\n",
    "# Calling the Natural Language Processing REST interface\n",
    "#\n",
    "rdd1 = sample.map(lambda x: SentimentAnalysis(x[1]))\n",
    "rdd2 =  rdd1.flatMap(lambda x: x['sentences'] )\\\n",
    "            .flatMap(lambda x: [(x['sentiment']['magnitude'], x['sentiment']['score'], [x['text']$\n",
    "results = rdd2.take(50)\n",
    "for item in results:\n",
    "  print 'Magnitude= ',item[0],' | Score= ',item[1], ' | Text= ',item[2]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example Dataproc ML Script [3]\n",
    "\n",
    "This program builds on the previous one. Instead of reading a poem it is going to read an entire book. However, it could just as easily read and process an entire library.\n",
    "\n",
    "Adds filter (in the pipeline) and sort (Python).\n",
    "\n",
    "This gives a list of the lines in the book with the strongest sentiment, both positive and negative.\n",
    "\n",
    "Now this was just a book. Imagine how you could use this to sort through social media commentary. For example, consider the feedback left by customers on a shopping website. You could use this kind of data analysis to identify the most admired and most despised products."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/usr/bin/env python\n",
    "#! 03-dataprocML.py\n",
    "# Copyright 2018 Google Inc.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "'''\n",
    "  This program reads a text file and passes to a Natural Language Processing\n",
    "  service, sentiment analysis, and processes the results in Spark.\n",
    "'''\n",
    "import logging\n",
    "import argparse\n",
    "import json\n",
    "import os\n",
    "from googleapiclient.discovery import build\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(\"local\", \"Simple App\")\n",
    "'''\n",
    "You must set these values for the job to run.\n",
    "'''\n",
    "APIKEY=\"removed_API_key\"   # CHANGE\n",
    "print APIKEY\n",
    "PROJECT_ID=\"qwiklabs-gcp-40459444aeb2e780\"  # CHANGE\n",
    "print PROJECT_ID\n",
    "BUCKET=\"qwiklabs-gcp-40459444aeb2e780\"   # CHANGE\n",
    "## Wrappers around the NLP REST interface\n",
    "def SentimentAnalysis(text):\n",
    "    from googleapiclient.discovery import build#!/usr/bin/env python\n",
    "    lservice = build('language', 'v1beta1', developerKey=APIKEY)\n",
    "    response = lservice.documents().analyzeSentiment(\n",
    "        body={\n",
    "            'document': {\n",
    "                'type': 'PLAIN_TEXT',\n",
    "                'content': text\n",
    "            }\n",
    "        }).execute()\n",
    "    return response\n",
    "## main\n",
    "# We could use sc.textFiles(...)\n",
    "#\n",
    "#   However, that will read each line of text as a separate object.\n",
    "#   And using the REST API to NLP for each line will rapidly exhaust the rate-limit quota \n",
    "#   producing HTTP 429 errors\n",
    "#\n",
    "#   Instead, it is more efficient to pass an entire document to NLP in a single call.\n",
    "#\n",
    "#   So we are using sc.wholeTextFiles(...)\n",
    "      This provides a file as a tuple.\n",
    "#      The first element is the file pathname, and second element is the content of the file.\n",
    "#\n",
    "sample = sc.wholeTextFiles(\"gs://{0}/sampledata/time-machine.txt\".format(BUCKET))\n",
    "# Calling the Natural Language Processing REST interface\n",
    "#\n",
    "# results = SentimentAnalysis(sampleline)\n",
    "rdd1 = sample.map(lambda x: SentimentAnalysis(x[1]))\n",
    "# The RDD contains a dictionary, using the key 'sentences' picks up each individual sentence\n",
    "# The value that is returned is a list. And inside the list is another dictionary\n",
    "# The key 'sentiment' produces a value of another list.\n",
    "# And the keys magnitude and score produce values of floating numbers. \n",
    "#\n",
    "rdd2 =  rdd1.flatMap(lambda x: x['sentences'] )\\\n",
    "            .flatMap(lambda x: [(x['sentiment']['magnitude'], x['sentiment']['score'], [x['text']$\n",
    "# First item in the list tuple is magnitude\n",
    "# Filter on only the statements with the most intense sentiments\n",
    "#\n",
    "                                                                                        \n",
    "#      This provides a file as a tuple.\n",
    "#      The first element is the file pathname, and second element is the content of the file.\n",
    "#\n",
    "sample = sc.wholeTextFiles(\"gs://{0}/sampledata/time-machine.txt\".format(BUCKET))\n",
    "# Calling the Natural Language Processing REST interface\n",
    "#\n",
    "# results = SentimentAnalysis(sampleline)\n",
    "rdd1 = sample.map(lambda x: SentimentAnalysis(x[1]))\n",
    "# The RDD contains a dictionary, using the key 'sentences' picks up each individual sentence\n",
    "# The value that is returned is a list. And inside the list is another dictionary\n",
    "# The key 'sentiment' produces a value of another list.\n",
    "# And the keys magnitude and score produce values of floating numbers. \n",
    "#\n",
    "rdd2 =  rdd1.flatMap(lambda x: x['sentences'] )\\\n",
    "            .flatMap(lambda x: [(x['sentiment']['magnitude'], x['sentiment']['score'], [x['text']$\n",
    "# First item in the list tuple is magnitude\n",
    "# Filter on only the statements with the most intense sentiments\n",
    "#\n",
    "rdd3 =  rdd2.filter(lambda x: x[0]>.75)\n",
    "results = sorted(rdd3.take(50))\n",
    "print '\\n\\n'\n",
    "for item in results:\n",
    "  print 'Magnitude= ',item[0],' | Score= ',item[1], ' | Text= ',item[2],'\\n'"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}