In [1]:
from pyspark import SparkContext
sc =SparkContext()
ListaPalavras = ['gato', 'elefante', 'rato', 'rato', 'gato']
palavrasRDD = sc.parallelize(ListaPalavras, 4)
print type(palavrasRDD)
map() para aplicar a transformação em cada palavra do RDD.str.format().help(). Vamos utilizar a padronização de documentação sugerida para o Python, manteremos essa documentação em inglês.
In [2]:
# EXERCICIO
def Plural(palavra):
"""Adds an 's' to `palavra`.
Args:
palavra (str): A string.
Returns:
str: A string with 's' added to it.
"""
return "{0}{1}".format(palavra,"s")#<COMPLETAR>
print Plural('gato')
In [3]:
help(Plural)
In [4]:
assert Plural('rato')=='ratos', 'resultado incorreto!'
print 'OK'
In [5]:
# EXERCICIO
pluralRDD = palavrasRDD.map(Plural)#<COMPLETAR>
print pluralRDD.collect()
In [6]:
assert pluralRDD.collect()==['gatos','elefantes','ratos','ratos','gatos'], 'valores incorretos!'
print 'OK'
In [8]:
# EXERCICIO
pluralLambdaRDD = palavrasRDD.map(lambda x: "{0}{1}".format(x,"s"))#<COMPLETAR>
print pluralLambdaRDD.collect()
In [9]:
assert pluralLambdaRDD.collect()==['gatos','elefantes','ratos','ratos','gatos'], 'valores incorretos!'
print 'OK'
In [12]:
# EXERCICIO
pluralTamanho = (pluralRDD.map(lambda x: len(x))
#<COMPLETAR>
).collect()
print pluralTamanho
In [13]:
assert pluralTamanho==[5,9,5,5,5], 'valores incorretos'
print "OK"
1 para cada palavra.(k,v) é chamada de RDD de tuplas ou pair RDD.map() com uma função lambda().
In [14]:
# EXERCICIO
palavraPar = palavrasRDD.map(lambda x: (x,1))#<COMPLETAR>
print palavraPar.collect()
In [15]:
assert palavraPar.collect() == [('gato',1),('elefante',1),('rato',1),('rato',1),('gato',1)], 'valores incorretos!'
print "OK"
groupByKey()
In [16]:
# EXERCICIO
palavrasGrupo = palavraPar.groupByKey()
for chave, valor in palavrasGrupo.collect():
print '{0}: {1}'.format(chave, list(valor))
In [19]:
assert sorted(palavrasGrupo.mapValues(lambda x: list(x)).collect()) == [('elefante', [1]), ('gato',[1, 1]), ('rato',[1, 1])], 'Valores incorretos!'
print "OK"
In [21]:
# EXERCICIO
contagemGroup = palavrasGrupo.mapValues(lambda x: sum (x))#<COMPLETAR>
print contagemGroup.collect()
In [22]:
assert list(sorted(contagemGroup.collect()))==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print "OK"
reduceByKey reduce() vista na aula anterior para os valores de cada chave. Dessa forma, a função de transformação pode ser aplicada em cada partição local para depois ser enviada para redistribuição de partições, reduzindo o total de dados sendo movidos e não mantendo listas grandes na memória.
In [26]:
# EXERCICIO
from operator import add
contagem = palavraPar.reduceByKey(add)#<COMPLETAR>
print contagem.collect()
In [27]:
assert sorted(contagem.collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print "OK"
In [34]:
# EXERCICIO
contagemFinal = (palavrasRDD.map(lambda x:(x,1)).reduceByKey(add)
#<COMPLETAR>
#<COMPLETAR>
)
contagemFinal = contagemFinal.collect()
print contagemFinal
In [35]:
assert sorted(contagemFinal)==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print "OK"
In [69]:
# EXERCICIO
palavrasUnicas = (palavrasRDD.map(lambda x:(x,1)).reduceByKey(lambda y,z:1)).collect()#<COMPLETAR>
palavrasUnicas = len(palavrasUnicas)
print palavrasUnicas
In [70]:
assert palavrasUnicas==3, 'valor incorreto!'
print "OK"
contagem.reduce() é aplicada em cada tupla do RDD. Para realizar a soma das contagens, primeiro é necessário mapear o RDD para um RDD contendo apenas os valores das frequências (sem as chaves).
In [90]:
# EXERCICIO
# add é equivalente a lambda x,y: x+y
palavrasRDD2 = sc.parallelize(contagemFinal)
#print palavrasRDD2.collect()
total = (palavrasRDD2.map(lambda x:(x[1])).reduce(add))
media = total / float(palavrasUnicas)
print total
print round(media, 2)
In [91]:
assert round(media, 2)==1.67, 'valores incorretos!'
print "OK"
contaPalavras
In [94]:
# EXERCICIO
def contaPalavras(chavesRDD):
"""Creates a pair RDD with word counts from an RDD of words.
Args:
chavesRDD (RDD of str): An RDD consisting of words.
Returns:
RDD of (str, int): An RDD consisting of (word, count) tuples.
"""
return (chavesRDD.map(lambda x: (x,1)).reduceByKey(add)
#<COMPLETAR>
#<COMPLETAR>
)
print contaPalavras(palavrasRDD).collect()
In [95]:
assert sorted(contaPalavras(palavrasRDD).collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print "OK"
removerPontuacao que converte todo o texto para minúscula, remove qualquer pontuação e espaços em branco no início ou final da palavra. Para isso, utilize a biblioteca re para remover todo texto que não seja letra, número ou espaço, encadeando com as funções de string para remover espaços em branco e converter para minúscula (veja Strings).
In [96]:
# EXERCICIO
import re
def removerPontuacao(texto):
"""Removes punctuation, changes to lower case, and strips leading and trailing spaces.
Note:
Only spaces, letters, and numbers should be retained. Other characters should should be
eliminated (e.g. it's becomes its). Leading and trailing spaces should be removed after
punctuation is removed.
Args:
texto (str): A string.
Returns:
str: The cleaned up string.
"""
return re.sub(r'[^A-Za-z0-9 ]', '', texto).strip().lower()
print removerPontuacao('Ola, quem esta ai??!')
print removerPontuacao(' Sem espaco e_sublinhado!')
In [97]:
assert removerPontuacao(' O uso de virgulas, embora permitido, nao deve contar. ')=='o uso de virgulas embora permitido nao deve contar', 'string incorreta!'
print "OK"
textFile() que recebe como entrada o nome do arquivo texto que queremos utilizar e o número de partições.removerPontuacao() para normalizar o texto e verificar as 15 primeiras linhas com o comando take().
In [108]:
# Apenas execute a célula
import os.path
import urllib2
url = 'http://www.gutenberg.org/cache/epub/100/pg100.txt' # url do livro
arquivo = os.path.join('Data','Aula02','shakespeare.txt') # local de destino: 'Data/Aula02/shakespeare.txt'
if os.path.isfile(arquivo): # verifica se já fizemos download do arquivo
print 'Arquivo já existe!'
else:
try:
response = urllib2.urlopen(url)
arquivo = (response.read()).split() #ja gera uma lista de palavras
except IOError:
print 'Impossível fazer o download: {0}'.format(url)
# lê o arquivo com textFile e aplica a função removerPontuacao
shakesRDD = (sc.textFile(arquivo).map(removerPontuacao))
# zipWithIndex gera tuplas (conteudo, indice) onde indice é a posição do conteudo na lista sequencial
# Ex.: sc.parallelize(['gato','cachorro','boi']).zipWithIndex() ==> [('gato',0), ('cachorro',1), ('boi',2)]
# sep.join() junta as strings de uma lista através do separador sep. Ex.: ','.join(['a','b','c']) ==> 'a,b,c'
print '\n'.join(shakesRDD
.zipWithIndex()
.map(lambda (linha, num): '{0}: {1}'.format(num,linha))
.take(15)
)
contaPalavras(), temos ainda que trabalhar em cima da nossa RDD:map() para gerar um novo RDD como uma lista de palavras.
In [119]:
# EXERCICIO
shakesPalavrasRDD = shakesRDD.map(lambda x: x.split())#<COMPLETAR>
total = shakesPalavrasRDD.count()
print shakesPalavrasRDD.take(5)
print total
map() gera uma lista para cada linha, criando um RDD contendo uma lista de listas.flatMap() que aplica a transformação do map(), porém achatando o retorno em forma de lista para uma lista unidimensional.
In [120]:
# EXERCICIO
shakesPalavrasRDD = shakesRDD.flatMap(lambda x: x.split())
total = shakesPalavrasRDD.count()
print shakesPalavrasRDD.top(5)
print total
Nota: os asserts abaixo de contagem de palavra podem falhar por diferença de formato do arquivo .txt antigo e novo. Eu avaliarei somente os códigos nesse trecho.
In [122]:
#assert total==927631 or total == 928908, "valor incorreto de palavras!"
#print "OK"
assert shakesPalavrasRDD.top(5)==[u'zwaggerd', u'zounds', u'zounds', u'zounds', u'zounds'],'lista incorreta de palavras'
print "OK"
In [129]:
# EXERCICIO
shakesLimpoRDD = shakesPalavrasRDD.filter(lambda x: len(x)>0)#<COMPLETAR>
total = shakesLimpoRDD.count()
print total
In [126]:
assert total==882996, 'valor incorreto!'
print "OK"
contaPalavras().takeOrdered para imprimir as 15 palavras mais frequentes.takeOrdered() pode receber um segundo parâmetro que instrui o Spark em como ordenar os elementos. Ex.:takeOrdered(15, key=lambda x: -x): ordem decrescente dos valores de x
In [140]:
# EXERCICIO
#print contaPalavras(shakesLimpoRDD).collect()
top15 = contaPalavras(shakesLimpoRDD).takeOrdered(15, lambda x: -x[1])#<COMPLETAR>
print '\n'.join(map(lambda (w, c): '{0}: {1}'.format(w, c), top15))
In [141]:
assert top15 == [(u'the', 27361), (u'and', 26028), (u'i', 20681), (u'to', 19150), (u'of', 17463),
(u'a', 14593), (u'you', 13615), (u'my', 12481), (u'in', 10956), (u'that', 10890),
(u'is', 9134), (u'not', 8497), (u'with', 7771), (u'me', 7769), (u'it', 7678)],'valores incorretos!'
print "OK"
In [142]:
import numpy as np
# Vamos criar uma função pNorm que recebe como parâmetro p e retorna uma função que calcula a pNorma
def pNorm(p):
"""Generates a function to calculate the p-Norm between two points.
Args:
p (int): The integer p.
Returns:
Dist: A function that calculates the p-Norm.
"""
def Dist(x,y):
return np.power(np.power(np.abs(x-y),p).sum(),1/float(p))
return Dist
In [143]:
# Vamos criar uma RDD com valores numéricos
numPointsRDD = sc.parallelize(enumerate(np.random.random(size=(10,100))))
In [172]:
# EXERCICIO
# Procure dentre os comandos do PySpark, um que consiga fazer o produto cartesiano da base com ela mesma
cartPointsRDD = numPointsRDD.cartesian(numPointsRDD)#<COMPLETAR>
# Aplique um mapa para transformar nossa RDD em uma RDD de tuplas ((id1,id2), (vetor1,vetor2))
# DICA: primeiro utilize o comando take(1) e imprima o resultado para verificar o formato atual da RDD
cartPointsParesRDD = cartPointsRDD.map(lambda ((x1,x2),(y1,y2)):((x1,y1),(x2,y2)))#<COMPLETAR>
#print cartPointsParesRDD
#Aplique um mapa para calcular a Distância Euclidiana entre os pares
Euclid = pNorm(2)
distRDD = cartPointsParesRDD.map(lambda ((x1,y1),(x2,y2)): Euclid(x2,y2))#<COMPLETAR>
#print(distRDD.collect())
# Encontre a distância máxima, mínima e média, aplicando um mapa que transforma (chave,valor) --> valor
# e utilizando os comandos internos do pyspark para o cálculo da min, max, mean
#statRDD = distRDD.<COMPLETAR>
#minv, maxv, meanv = statRDD.<COMPLETAR>, statRDD.<COMPLETAR>, statRDD.<COMPLETAR>
minv, maxv, meanv = distRDD.min(), distRDD.max(), distRDD.mean()#<COMPLETAR>
print minv, maxv, meanv
In [173]:
assert (minv.round(2), maxv.round(2), meanv.round(2))==(0.0, 4.70, 3.65), 'Valores incorretos'
print "OK"
In [174]:
# Vamos criar uma função para calcular a distância de Hamming
def Hamming(x,y):
"""Calculates the Hamming distance between two binary vectors.
Args:
x, y (np.array): Array of binary integers x and y.
Returns:
H (int): The Hamming distance between x and y.
"""
return (x!=y).sum()
# Vamos criar uma função para calcular a distância de Jaccard
def Jaccard(x,y):
"""Calculates the Jaccard distance between two binary vectors.
Args:
x, y (np.array): Array of binary integers x and y.
Returns:
J (int): The Jaccard distance between x and y.
"""
return (x==y).sum()/float( np.maximum(x,y).sum() )
In [175]:
# Vamos criar uma RDD com valores categóricos
catPointsRDD = sc.parallelize(enumerate([['alto', 'caro', 'azul'],
['medio', 'caro', 'verde'],
['alto', 'barato', 'azul'],
['medio', 'caro', 'vermelho'],
['baixo', 'barato', 'verde'],
]))
In [198]:
# EXERCICIO
# Crie um RDD de chaves únicas utilizando flatMap
chavesRDD = catPointsRDD.flatMap(lambda x: (x[1])).distinct()#.zipWithIndex()
chaves = dict((v,k) for k,v in enumerate(chavesRDD.collect()))
nchaves = len(chaves)
print chaves, nchaves
In [199]:
assert chaves=={'alto': 0, 'medio': 1, 'baixo': 2, 'barato': 3, 'azul': 4, 'verde': 5, 'caro': 6, 'vermelho': 7}, 'valores incorretos!'
print "OK"
assert nchaves==8, 'número de chaves incorreta'
print "OK"
In [200]:
def CreateNP(atributos,chaves):
"""Binarize the categorical vector using a dictionary of keys.
Args:
atributos (list): List of attributes of a given object.
chaves (dict): dictionary with the relation attribute -> index
Returns:
array (np.array): Binary array of attributes.
"""
array = np.zeros(len(chaves))
for atr in atributos:
array[ chaves[atr] ] = 1
return array
# Converte o RDD para o formato binário, utilizando o dict chaves
binRDD = catPointsRDD.map(lambda rec: (rec[0],CreateNP(rec[1], chaves)))
binRDD.collect()
Out[200]:
In [201]:
# EXERCICIO
# Procure dentre os comandos do PySpark, um que consiga fazer o produto cartesiano da base com ela mesma
cartBinRDD = binRDD.cartesian(binRDD)#<COMPLETAR>
# Aplique um mapa para transformar nossa RDD em uma RDD de tuplas ((id1,id2), (vetor1,vetor2))
# DICA: primeiro utilize o comando take(1) e imprima o resultado para verificar o formato atual da RDD
cartBinParesRDD = cartBinRDD.map(lambda ((x1,x2),(y1,y2)):((x1,y1),(x2,y2)))#<COMPLETAR>
distRDD = cartPointsParesRDD.map(lambda ((x1,y1),(x2,y2)): Euclid(x2,y2))#<COMPLETAR>
# Aplique um mapa para calcular a Distância de Hamming e Jaccard entre os pares
hamRDD = cartBinParesRDD.map(lambda ((x1,y1),(x2,y2)): Hamming(x2,y2))#<COMPLETAR>
jacRDD = cartBinParesRDD.map(lambda ((x1,y1),(x2,y2)): Jaccard(x2,y2))#<COMPLETAR>
# Encontre a distância máxima, mínima e média, aplicando um mapa que transforma (chave,valor) --> valor
# e utilizando os comandos internos do pyspark para o cálculo da min, max, mean
#statHRDD = hamRDD.<COMPLETAR>
#statJRDD = jacRDD.<COMPLETAR>
Hmin, Hmax, Hmean = hamRDD.min(), hamRDD.max(), hamRDD.mean()
Jmin, Jmax, Jmean = jacRDD.min(), jacRDD.max(), jacRDD.mean()
#Hmin, Hmax, Hmean = statHRDD.<COMPLETAR>, statHRDD.<COMPLETAR>, statHRDD.<COMPLETAR>
#Jmin, Jmax, Jmean = statJRDD.<COMPLETAR>, statJRDD.<COMPLETAR>, statJRDD.<COMPLETAR>
print "\t\tMin\tMax\tMean"
print "Hamming:\t{:.2f}\t{:.2f}\t{:.2f}".format(Hmin, Hmax, Hmean )
print "Jaccard:\t{:.2f}\t{:.2f}\t{:.2f}".format( Jmin, Jmax, Jmean )
In [202]:
assert (Hmin.round(2), Hmax.round(2), Hmean.round(2)) == (0.00,6.00,3.52), 'valores incorretos'
print "OK"
assert (Jmin.round(2), Jmax.round(2), Jmean.round(2)) == (0.33,2.67,1.14), 'valores incorretos'
print "OK"
In [ ]: