<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Trending Topics de Twitter con Spark Streaming
Este código no tiene la calidad requerida para sistemas en producción
El Broker de Twitter (read.py) usa la librería socketstream y queda a la escucha en un socket TCP (como servidor). El script a su vez está filtrando los tweets provenientes de la Streaming API de Twitter
Step1:  Crear el contexto de Streaming. Con la configuración
Step2:  Una vez instanciados los contextos, nos conectamos a la fuente de datos
Step3:  Soporte de Window Operations y tuplas para conteo de etiquetas
Creamos una ventana de tiempo.
El RDD obtenido del DStream se creará cada 10 segundos, pero los datos que estarán en el RDD serán por los últimos 20 segundos.
Spark Streaming
Step4:  Creamos un objeto namedtuple para almacenar las etiquetas y sus conteos.
Step5:  Procesar el flujo de datos
Cada tweet/mensaje que se recibe, será cortado en palabras (flatMap()),
Por cada registro generado, se filtrarán  las palabras que empiezan con el simbolo #,
Después se convertirán esos registros a minúsculas, 
Se ejecuta una acción de mapeo a (palabra, 1)
Luego, se reducen los pares (llave, valor) y cuentan las ocurrencias de cada Hashtag (palabra, n)
El flujo se convierte en un Dataframe con las etiquetas resultantes y sus conteos.
Entonces, se ordenan las etiquetas de forma descendente,
Y se toman los primeros 10 registos,
Finalmente, esos 10 registros se almacenan en una tabla temporal
Step6:  Iniciar el proceso de Spark Streaming
Step7:  Graficar las tendencias
Podemos cambiar los valores de time.sleep() junto con la operación en ventana de tiempo para actualizaciones más cortas
<ASSISTANT_TASK:>
Python Code:
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext
from pyspark.sql.functions import desc
from collections import namedtuple
# Namedtuple: https://pymotw.com/2/collections/namedtuple.html, http://stackoverflow.com/questions/2970608/what-are-named-tuples-in-python
import time
import matplotlib.pyplot as plt
import seaborn as sn
%matplotlib inline
from IPython import display
Explanation: Trending Topics de Twitter con Spark Streaming
Este código no tiene la calidad requerida para sistemas en producción
El Broker de Twitter (read.py) usa la librería socketstream y queda a la escucha en un socket TCP (como servidor). El script a su vez está filtrando los tweets provenientes de la Streaming API de Twitter: https://dev.twitter.com/streaming/overview
A partir de este momento, el broker puede ser ejecutado.
Dependencias y configuración de contextos
Cargar las librerías para el contexto: de Spark, SQL y Streaming. Así como otras dependencias del ejercicio
End of explanation
sc = SparkContext("local[2]", "TwitterTrend")
ssc = StreamingContext(sc, 10)
sqlContext = SQLContext(sc)
Explanation: Crear el contexto de Streaming. Con la configuración: "TwitterTrend" como nombre del programa, 10 segundos como intervalo para batch y dos hilos de ejecución.
De igual manera, crear la instancia del contexto de SQL en Spark.
End of explanation
socket_stream = ssc.socketTextStream("localhost", 5555)
Explanation: Una vez instanciados los contextos, nos conectamos a la fuente de datos:
End of explanation
lines = socket_stream.window( 20 )
Explanation: Soporte de Window Operations y tuplas para conteo de etiquetas
Creamos una ventana de tiempo.
El RDD obtenido del DStream se creará cada 10 segundos, pero los datos que estarán en el RDD serán por los últimos 20 segundos.
Spark Streaming: Window Operations
End of explanation
fields = ("tag", "count" )
Tweet = namedtuple( 'Tweet', fields )
Explanation: Creamos un objeto namedtuple para almacenar las etiquetas y sus conteos.
End of explanation
( lines.flatMap( lambda text: text.split( " " ) )
  .filter( lambda word: word.lower().startswith("#") )
  .map( lambda word: ( word.lower(), 1 ) )
  .reduceByKey( lambda a, b: a + b )
  .map( lambda rec: Tweet( rec[0], rec[1] ) )
  .foreachRDD( lambda rdd: rdd.toDF().sort( desc("count") )
              .limit(10).registerTempTable("tweets") ) )
Explanation: Procesar el flujo de datos
Cada tweet/mensaje que se recibe, será cortado en palabras (flatMap()),
Por cada registro generado, se filtrarán  las palabras que empiezan con el simbolo #,
Después se convertirán esos registros a minúsculas, 
Se ejecuta una acción de mapeo a (palabra, 1)
Luego, se reducen los pares (llave, valor) y cuentan las ocurrencias de cada Hashtag (palabra, n)
El flujo se convierte en un Dataframe con las etiquetas resultantes y sus conteos.
Entonces, se ordenan las etiquetas de forma descendente,
Y se toman los primeros 10 registos,
Finalmente, esos 10 registros se almacenan en una tabla temporal
End of explanation
# Lazy evaluation
ssc.start() 
Explanation: Iniciar el proceso de Spark Streaming
End of explanation
for i in range(0,1000):
  time.sleep( 1 )
  top_10_tweets = sqlContext.sql( 'Select tag, count from tweets' )
  top_10_df = top_10_tweets.toPandas()
  display.clear_output(wait=True)
  plt.figure( figsize = ( 10, 8 ) )
  sn.barplot( x="count", y="tag", data=top_10_df)
  plt.show()
ssc.stop()
Explanation: Graficar las tendencias
Podemos cambiar los valores de time.sleep() junto con la operación en ventana de tiempo para actualizaciones más cortas
End of explanation
<END_TASK>