<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <h2>Establish environment variables and services for Google Cloud API access</h2>
In Console, go to <b> Products & services </b> > <b> APIs & services </b> > <b> Credentials </b>
Click on <b> Create Credentials </b> and select <b> API Key </b>
Copy the API KEy and paste it in the APIKEY field below.
+
In Console, go to <b> Products & services </b> > <b> Home </b>
Select and copy the Project ID
Paste the Project ID into both the PROJECT_ID field and the BUCKET field below.
+
Run the next block
Step1:  <h2> Define an API calling function </h2>
Run the following block of code to define a language service API interface
When this is called it will pass the text to the service using a JSON formatted block
And it will receive a JSON response from the Google Cloud Language service
The response will be automatically represented by Python as a 'dict' object (a dictionary)
Step2:  <h2> Test the Sentiment Analysis </h2>
Use a simple string to test the function and verify all the API elements are working.
Step3:  <h2>Use the Dataproc cluster to run a Spark job that uses the Machine Learning API </h2>
Step4:  <h2> Working with the results in Python </h2>
The collect() action is good for validating sample output, but don't use it on big data because all the results must fit in memory.
The following shows how the data that is passed back is formatted as a Python list of dictionary objects.
Step5:  <h2> Using another feature of the Natural Language API </h2>
In this code block there is an analyze Entities version of the Language Service
Analysis of entities identifies and recognizes items in text
Step6:  <h2> Working with the results in Spark </h2>
Working with results in Python can be useful. However, if the results are too big to fit in memory, you will want to perform transformations
on the data while still in RDDs using Spark. In this section you will explore transforming the results of the TailoredAnalysis function in Spark.
<h3> Load some sample files into Cloud Storage </h3>
Step 1
Step7:  <h2> Save as text to Cloud Storage </h2>
<h3>Write file to Cloud Storage</h3>
Replace the bucket in the example with your bucket.
Run the next block to save the RDD to cloud storage. 
repartition(1) reorganizes the RDD internally into a single partition.
saveAsTextFile saved the partition in a folder called sampleoutput. 
<h3>View Cloud Storage in Console</h3>
In Console, go to the Cloud Storage Browser, locate the sampleoutput folder, and look inside.
Inside the folder you will find part-xxxxx coresponding to the partitions in the RDD.
<ASSISTANT_TASK:>
Python Code:
APIKEY="AIzaSyBQrrl4SZhE3QtxsnbjY2WTdgcBz0G0Rfs"   # CHANGE
print APIKEY
PROJECT_ID = "qwiklabs-gcp-14067121d7b1d12c"  # CHANGE
print PROJECT_ID 
BUCKET = "qwiklabs-gcp-14067121d7b1d12c"   # CHANGE
import os
os.environ['BUCKET'] = BUCKET
os.environ['PROJECT'] = PROJECT_ID
from googleapiclient.discovery import build
print("\n","Google Cloud API Client credentials established")
Explanation: <h2>Establish environment variables and services for Google Cloud API access</h2>
In Console, go to <b> Products & services </b> > <b> APIs & services </b> > <b> Credentials </b>
Click on <b> Create Credentials </b> and select <b> API Key </b>
Copy the API KEy and paste it in the APIKEY field below.
+
In Console, go to <b> Products & services </b> > <b> Home </b>
Select and copy the Project ID
Paste the Project ID into both the PROJECT_ID field and the BUCKET field below.
+
Run the next block
End of explanation
def SentimentAnalysis(text):
    from googleapiclient.discovery import build
    lservice = build('language', 'v1beta1', developerKey=APIKEY)
    response = lservice.documents().analyzeSentiment(
        body={
            'document': {
                'type': 'PLAIN_TEXT',
                'content': text
            }
        }).execute()
    
    return response
print("\n","Sentiment Analysis function defined.")
Explanation: <h2> Define an API calling function </h2>
Run the following block of code to define a language service API interface
When this is called it will pass the text to the service using a JSON formatted block
And it will receive a JSON response from the Google Cloud Language service
The response will be automatically represented by Python as a 'dict' object (a dictionary)
End of explanation
sampleline = u'There are places I remember, all my life though some have changed.'
results = SentimentAnalysis(sampleline)
print("\n","This is the Python object that is returned; a dictionary.")
print("\n")
print("Function returns :",type(results))
print(results)
import json
print("\n","This is the JSON formatted version of the object")
print(json.dumps(results, sort_keys=True, indent=4))
Explanation: <h2> Test the Sentiment Analysis </h2>
Use a simple string to test the function and verify all the API elements are working.
End of explanation
# Working with the smaller sample file
#
lines = sc.textFile("/sampledata/road-not-taken.txt")
#
# The Spark map transformation will execute SentimentAnalysis on each element in lines and store the result in sentiment.
# Remember that due to lazy execution, this line just queues up the transformation, it does not run yet.
# So you will not see errors at this point.
#
sentiment = lines.map(SentimentAnalysis)
#
#
print (type(sentiment))
# sentiment is a pyspark.rdd.PipelinedRDD
#
# If it is properly formed then an action such as sentiment.collect() will run the job.
# If not properly formed, it will throw errors.
#
output = sentiment.collect()
#
# The sentiment rdd contains JSON returns. In python these are collected into a list of dictionaries.
#
print(type(output))
print("\n")
for line in output:
  print(line)
Explanation: <h2>Use the Dataproc cluster to run a Spark job that uses the Machine Learning API </h2>
End of explanation
# 
# Ouput is a list of dictionaries
# When the list is iterated, each line is one dictionary 
# And the dictionary is double-subscripted
#
for line in output:
  print("Score: ",line['documentSentiment']['score'], " Magnitude :",line['documentSentiment']['magnitude'])
Explanation: <h2> Working with the results in Python </h2>
The collect() action is good for validating sample output, but don't use it on big data because all the results must fit in memory.
The following shows how the data that is passed back is formatted as a Python list of dictionary objects.
End of explanation
def TailoredAnalysis(text):
    from googleapiclient.discovery import build
    lservice = build('language', 'v1beta1', developerKey=APIKEY)
    response = lservice.documents().analyzeEntities(
        body={
            'document': {
                'type': 'PLAIN_TEXT',
                'content': text
            }
        }).execute()
    return response
  
print("\n","Tailored Analysis function defined.")
Explanation: <h2> Using another feature of the Natural Language API </h2>
In this code block there is an analyze Entities version of the Language Service
Analysis of entities identifies and recognizes items in text
End of explanation
# [STEP 1] HDFS 
#lines = sc.textFile("/sampledata/road-not-taken.txt")
#
#
# [STEP 2] Cloud Storage 
#lines = sc.textFile("gs://<your-bucket>/time-machine-P1.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P2.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P3.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P4.txt")
lines = sc.textFile("gs://qwiklabs-gcp-14067121d7b1d12c/time-machine-P1.txt")
#
#
#
entities = lines.map(TailoredAnalysis)
from operator import add
rdd = entities.map(lambda x: x['entities'])
#
# results = rdd.flatMap(lambda x: x ).filter(lambda x: x['type']==u'PERSON').map(lambda x:(x['name'],1)).reduceByKey(add)
#
# It is common practice to use line continuation "\" to help make the chain more readable
results  = rdd.flatMap(lambda x: x )\
              .filter(lambda x: x['type']==u'PERSON')\
              .map(lambda x:(x['name'],1))\
              .reduceByKey(add)
              
print(sorted(results.take(25)))
# [STEP 3] Cloud Storage 
#lines = sc.textFile("gs://<your-bucket>/time-machine-P1.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P2.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P3.txt")
#lines = sc.textFile("gs://<your-bucket>/time-machine-P4.txt")
#
lines = sc.textFile("gs://qwiklabs-gcp-14067121d7b1d12c/time-machine-P2.txt")
#
entities = lines.map(TailoredAnalysis)
from operator import add
rdd = entities.map(lambda x: x['entities'])
#
# results = rdd.flatMap(lambda x: x ).filter(lambda x: x['type']==u'PERSON').map(lambda x:(x['name'],1)).reduceByKey(add)
#
# It is common practice to use line continuation "\" to help make the chain more readable
results  = rdd.flatMap(lambda x: x )\
              .filter(lambda x: x['type']==u'LOCATION')\
              .map(lambda x:(x['name'],1))\
              .reduceByKey(add)
              
print(sorted(results.take(25)))
Explanation: <h2> Working with the results in Spark </h2>
Working with results in Python can be useful. However, if the results are too big to fit in memory, you will want to perform transformations
on the data while still in RDDs using Spark. In this section you will explore transforming the results of the TailoredAnalysis function in Spark.
<h3> Load some sample files into Cloud Storage </h3>
Step 1: Locate some samle text such as a news article in your browser. Copy the text into a file called sample.txt
Step 2: SSH to the Master Node
Step 3: Upload the file to Cloud Storage
gsutil cp sample.txt gs://[your-bucket]
Step 4: Download some more prepared sample files: 
curl https://storage.googleapis.com/cloud-training/archdp/sherlock2.txt > sherlock2.txt
curl https://storage.googleapis.com/cloud-training/archdp/sherlock3.txt > sherlock3.txt
curl https://storage.googleapis.com/cloud-training/archdp/sherlock4.txt > sherlock4.txt
Step 5: Upload the files to your Cloud Storage:  
gsutil cp sample* gs://[your-bucket]
<h3> Run the analysis and variations </h3>
This code recognizes people and locations mentioned in the document, and returns a list of them with the number of mentions.
+
Step 6: In the following code block, replace the bucket name [your-bucket] with your bucket
After you run the block and see the results, change the word 'PERSON' to 'LOCATION' and run it again
End of explanation
# Replace with your bucket
#
results.repartition(1).saveAsTextFile("gs://qwiklabs-gcp-14067121d7b1d12c/sampleoutput/")
print("Output to Cloud Storage is complete.")
Explanation: <h2> Save as text to Cloud Storage </h2>
<h3>Write file to Cloud Storage</h3>
Replace the bucket in the example with your bucket.
Run the next block to save the RDD to cloud storage. 
repartition(1) reorganizes the RDD internally into a single partition.
saveAsTextFile saved the partition in a folder called sampleoutput. 
<h3>View Cloud Storage in Console</h3>
In Console, go to the Cloud Storage Browser, locate the sampleoutput folder, and look inside.
Inside the folder you will find part-xxxxx coresponding to the partitions in the RDD.
End of explanation
<END_TASK>