notebook.community

Edit and run

Spark + Python = PySpark

Esse notebook introduz os conceitos básicos do Spark através de sua interface com a linguagem Python. Como aplicação inicial faremos o clássico examplo de contador de palavras . Com esse exemplo é possível entender a lógica de programação funcional para as diversas tarefas de exploração de dados distribuídos.

Para isso utilizaremos o livro texto Trabalhos completos de William Shakespeare obtidos do Projeto Gutenberg. Veremos que esse mesmo algoritmo pode ser empregado em textos de qualquer tamanho.

Esse notebook contém:

Parte 1: Criando uma base RDD e RDDs de tuplas

Parte 2: Manipulando RDDs de tuplas

Parte 3: Encontrando palavras únicas e calculando médias

Parte 4: Aplicar contagem de palavras em um arquivo

Parte 5: Similaridade entre Objetos

Para os exercícios é aconselhável consultar a documentação da API do PySpark

Part 1: Criando e Manipulando RDDs

Nessa parte do notebook vamos criar uma base RDD a partir de uma lista com o comando parallelize.

(1a) Criando uma base RDD

Podemos criar uma base RDD de diversos tipos e fonte do Python com o comando sc.parallelize(fonte, particoes), sendo fonte uma variável contendo os dados (ex.: uma lista) e particoes o número de partições para trabalhar em paralelo.





In [ ]:



    


ListaPalavras = ['gato', 'elefante', 'rato', 'rato', 'gato']
palavrasRDD = sc.parallelize(ListaPalavras, 4)
print(type(palavrasRDD))

(1b) Plural

Vamos criar uma função que transforma uma palavra no plural adicionando uma letra 's' ao final da string. Em seguida vamos utilizar a função map() para aplicar a transformação em cada palavra do RDD.

Em Python (e muitas outras linguagens) a concatenação de strings é custosa. Uma alternativa melhor é criar uma nova string utilizando str.format().

Nota: a string entre os conjuntos de três aspas representa a documentação da função. Essa documentação é exibida com o comando help(). Vamos utilizar a padronização de documentação sugerida para o Python, manteremos essa documentação em inglês.





In [ ]:



    


# EXERCICIO
def Plural(palavra):
    """Adds an 's' to `palavra`.

    Args:
        palavra (str): A string.

    Returns:
        str: A string with 's' added to it.
    """
    return <COMPLETAR>

print(Plural('gato'))



    











In [ ]:



    


help(Plural)



    











In [ ]:



    


assert Plural('rato')=='ratos', 'resultado incorreto!'
print ('OK')

(1c) Aplicando a função ao RDD

Transforme cada palavra do nosso RDD em plural usando map()

Em seguida, utilizaremos o comando collect() que retorna a RDD como uma lista do Python.





In [ ]:



    


# EXERCICIO
pluralRDD = palavrasRDD.<COMPLETAR>
print (pluralRDD.collect())



    











In [ ]:



    


assert pluralRDD.collect()==['gatos','elefantes','ratos','ratos','gatos'], 'valores incorretos!'
print ('OK')

Nota: utilize o comando collect() apenas quando tiver certeza de que a lista caberá na memória. Para gravar os resultados de volta em arquivo texto ou base de dados utilizaremos outro comando.

(1d) Utilizando uma função lambda

Repita a criação de um RDD de plurais, porém utilizando uma função lambda.





In [ ]:



    


# EXERCICIO
pluralLambdaRDD = palavrasRDD.<COMPLETAR>
print (pluralLambdaRDD.collect())



    











In [ ]:



    


assert pluralLambdaRDD.collect()==['gatos','elefantes','ratos','ratos','gatos'], 'valores incorretos!'
print ('OK')

(1e) Tamanho de cada palavra

Agora use map() e uma função lambda para retornar o número de caracteres em cada palavra. Utilize collect() para armazenar o resultado em forma de listas na variável destino.





In [ ]:



    


# EXERCICIO
pluralTamanho = (pluralRDD
                 <COMPLETAR>
                 .collect()
                 )
print (pluralTamanho)



    











In [ ]:



    


assert pluralTamanho==[5,9,5,5,5], 'valores incorretos'
print ("OK")

(1f) RDDs de pares e tuplas

Para contar a frequência de cada palavra de maneira distribuída, primeiro devemos atribuir um valor para cada palavra do RDD. Isso irá gerar um base de dados (chave, valor). Desse modo podemos agrupar a base através da chave, calculando a soma dos valores atribuídos. No nosso caso, vamos atribuir o valor 1 para cada palavra.

Um RDD contendo a estrutura de tupla chave-valor (k,v) é chamada de RDD de tuplas ou pair RDD.

Vamos criar nosso RDD de pares usando a transformação map() com uma função lambda().





In [ ]:



    


# EXERCICIO
palavraPar = palavrasRDD.<COMPLETAR>
print (palavraPar.collect())



    











In [ ]:



    


assert palavraPar.collect() == [('gato',1),('elefante',1),('rato',1),('rato',1),('gato',1)], 'valores incorretos!'
print ("OK")

Parte 2: Manipulando RDD de tuplas

Vamos manipular nossa RDD para contar as palavras do texto.

(2a) Função groupByKey()

A função groupByKey() agrupa todos os valores de um RDD através da chave (primeiro elemento da tupla) agregando os valores em uma lista.

Essa abordagem tem um ponto fraco pois:

  • #### A operação requer que os dados distribuídos sejam movidos em massa para que permaneçam na partição correta.
  • #### As listas podem se tornar muito grandes. Imagine contar todas as palavras do Wikipedia: termos comuns como "a", "e" formarão uma lista enorme de valores que pode não caber na memória do processo escravo.




In [ ]:



    


# EXERCICIO
palavrasGrupo = palavraPar.groupByKey()
for chave, valor in palavrasGrupo.collect():
    valores = list(valor)
    print(f'{chave}: {valores}')



    











In [ ]:



    


assert sorted(palavrasGrupo.mapValues(lambda x: list(x)).collect()) == [('elefante', [1]), ('gato',[1, 1]), ('rato',[1, 1])], 'Valores incorretos!'
print ("OK")

(2b) Calculando as contagens

Após o groupByKey() nossa RDD contém elementos compostos da palavra, como chave, e um iterador contendo todos os valores correspondentes aquela chave.

Utilizando a transformação map() e a função sum(), contrua um novo RDD que consiste de tuplas (chave, soma).





In [ ]:



    


# EXERCICIO
contagemGroup = palavrasGrupo.<COMPLETAR>
print (contagemGroup.collect())



    











In [ ]:



    


assert sorted(contagemGroup.collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print ("OK")

(2c) reduceByKey

Um comando mais interessante para a contagem é o reduceByKey() que cria uma nova RDD de tuplas.

Essa transformação aplica a transformação 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 [ ]:



    


# EXERCICIO
contagem = palavraPar.<COMPLETAR>
print( contagem.collect())



    











In [ ]:



    


assert sorted(contagem.collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print ("OK")

(2d) Agrupando os comandos

A forma mais usual de realizar essa tarefa, partindo do nosso RDD palavrasRDD, é encadear os comandos map e reduceByKey em uma linha de comando.





In [ ]:



    


# EXERCICIO
contagemFinal = (palavrasRDD
                 <COMPLETAR>
                 <COMPLETAR>                 
                 )
print (contagemFinal.collect())



    











In [ ]:



    


assert sorted(contagemFinal.collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print ("OK")

Parte 3: Encontrando as palavras únicas e calculando a média de contagem

(3a) Palavras Únicas

Calcule a quantidade de palavras únicas do RDD. Utilize o mesmo pipeline anterior porém substituindo .collect() por outro método do API da RDD.





In [ ]:



    


# EXERCICIO
palavrasUnicas = (contagemFinal
                  <COMPLETAR>
                 )
print (palavrasUnicas)



    











In [ ]:



    


assert palavrasUnicas==3, 'valor incorreto!'
print ("OK")

(3b) Calculando a Média de contagem de palavras

Encontre a média de frequência das palavras utilizando o RDD contagem.

Note que a função do comando 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 [ ]:



    


# EXERCICIO
# add é equivalente a lambda x,y: x+y
from operator import add
total = (contagemFinal
         <COMPLETAR>
         <COMPLETAR>
         )
media = total / float(palavrasUnicas)
print (total)
print (round(media, 2))



    











In [ ]:



    


assert round(media, 2)==1.67, 'valores incorretos!'
print ("OK")

Parte 4: Aplicar nosso algoritmo em um arquivo

(4a) Função contaPalavras

Para podermos aplicar nosso algoritmo genéricamente em diversos RDDs, vamos primeiro criar uma função para aplicá-lo em qualquer fonte de dados. Essa função recebe de entrada um RDD contendo uma lista de chaves (palavras) e retorna um RDD de tuplas com as chaves e a contagem delas nessa RDD





In [ ]:



    


# 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
            <COMPLETAR>
            <COMPLETAR>
           )

print (contaPalavras(palavrasRDD).collect())



    











In [ ]:



    


assert sorted(contaPalavras(palavrasRDD).collect())==[('elefante',1), ('gato',2), ('rato',2)], 'valores incorretos!'
print ("OK")

(4b) Normalizando o texto

Quando trabalhamos com dados reais, geralmente precisamos padronizar os atributos de tal forma que diferenças sutis por conta de erro de medição ou diferença de normatização, sejam desconsideradas. Para o próximo passo vamos padronizar o texto para:

  • #### Padronizar a capitalização das palavras (tudo maiúsculo ou tudo minúsculo).
  • #### Remover pontuação.
  • #### Remover espaços no início e no final da palavra.

Crie uma função 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 [ ]:



    


# 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 [ ]:



    


assert removerPontuacao(' O uso de virgulas, embora permitido, nao deve contar. ')=='o uso de virgulas embora permitido nao deve contar', 'string incorreta!'
print ("OK")

(4c) Carregando arquivo texto

Para a próxima parte vamos utilizar o livro Trabalhos completos de William Shakespeare do Projeto Gutenberg.

Para converter um texto em uma RDD, utilizamos a função textFile() que recebe como entrada o nome do arquivo texto que queremos utilizar e o número de partições.

O nome do arquivo texto pode se referir a um arquivo local ou uma URI de arquivo distribuído (ex.: hdfs://).

Vamos também aplicar a função removerPontuacao() para normalizar o texto e verificar as 15 primeiras linhas com o comando take().





In [ ]:



    


# Apenas execute a célula
import os.path

arquivo = os.path.join('Data', 'pg100.txt') 

# lê o arquivo com textFile e aplica a função removerPontuacao        
shakesRDD = (sc
             .textFile(arquivo, 8)
             .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: '{0}: {1}'.format(linha[0],linha[1]))
                .take(15)
               ))

(4d) Extraindo as palavras

Antes de poder usar nossa função contaPalavras(), temos ainda que trabalhar em cima da nossa RDD:

  • #### Precisamos gerar listas de palavras ao invés de listas de sentenças.
  • #### Eliminar linhas vazias.

As strings em Python tem o método split() que faz a separação de uma string por separador. No nosso caso, queremos separar as strings por espaço.

Utilize a função map() para gerar um novo RDD como uma lista de palavras.





In [ ]:



    


# EXERCICIO
shakesPalavrasRDD = shakesRDD.<COMPLETAR>
total = shakesPalavrasRDD.count()
print (shakesPalavrasRDD.take(5))
print (total)

Conforme deve ter percebido, o uso da função map() gera uma lista para cada linha, criando um RDD contendo uma lista de listas.

Para resolver esse problema, o Spark possui uma função análoga chamada flatMap() que aplica a transformação do map(), porém achatando o retorno em forma de lista para uma lista unidimensional.





In [ ]:



    


# EXERCICIO
shakesPalavrasRDD = shakesRDD.flatMap(lambda x: x.split())
total = shakesPalavrasRDD.count()
print (shakesPalavrasRDD.take(5))
print (total)



    











In [ ]:



    


assert total==959359, "valor incorreto de palavras!"
print ("OK")
assert shakesPalavrasRDD.take(5)==['project', 'gutenbergs', 'the', 'complete', 'works'],'lista incorreta de palavras'
print ("OK")

Reparem que o flatMap já eliminou as entradas vazias.

(4f) Contagem de palavras

Agora que nossa RDD contém uma lista de palavras, podemos aplicar nossa função contaPalavras().

Aplique a função em nossa RDD e utilize a função 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 [ ]:



    


# EXERCICIO
top15 = <COMPLETAR>
print ('\n'.join(map(lambda x: f'{x[0]}: {x[1]}', top15)))



    











In [ ]:



    


assert top15 == [('the', 29996), ('and', 28353), ('i', 21860), ('to', 20885), ('of', 18811), ('a', 15992), ('you', 14439), ('my', 13191), ('in', 12027), ('that', 11782), ('is', 9711), ('not', 9068), ('with', 8521), ('me', 8271), ('for', 8184)],'valores incorretos!'
print ("OK")

Parte 5: Similaridade entre Objetos

Nessa parte do laboratório vamos aprender a calcular a distância entre atributos numéricos, categóricos e textuais.

(5a) Vetores no espaço Euclidiano

Quando nossos objetos são representados no espaço Euclidiano, medimos a similaridade entre eles através da p-Norma definida por:

$$d(x,y,p) = (\sum_{i=1}^{n}{|x_i - y_i|^p})^{1/p}$$

As normas mais utilizadas são $p=1,2,\infty$ que se reduzem em distância absoluta, Euclidiana e máxima distância:

$$d(x,y,1) = \sum_{i=1}^{n}{|x_i - y_i|}$$

$$d(x,y,2) = (\sum_{i=1}^{n}{|x_i - y_i|^2})^{1/2}$$

$$d(x,y,\infty) = \max(|x_1 - y_1|,|x_2 - y_2|, ..., |x_n - y_n|)$$





In [ ]:



    


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 [ ]:



    


# Vamos criar uma RDD com valores numéricos
np.random.seed(42)
numPointsRDD = sc.parallelize(enumerate(np.random.random(size=(10,100))))



    











In [ ]:



    


# EXERCICIO
# Procure dentre os comandos do PySpark, um que consiga fazer o produto cartesiano da base com ela mesma
cartPointsRDD = 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.<COMPLETAR>


# Aplique um mapa para calcular a Distância Euclidiana entre os pares
Euclid = pNorm(2)
distRDD = cartPointsParesRDD.<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
statRDD = distRDD.<COMPLETAR>

minv, maxv, meanv = statRDD.<COMPLETAR>, statRDD.<COMPLETAR>, statRDD.<COMPLETAR>
print (minv, maxv, meanv)



    











In [ ]:



    


assert (minv.round(2), maxv.round(2), meanv.round(2))==(0.0, 4.71, 3.75), 'Valores incorretos'
print ("OK")

(5b) Valores Categóricos

Quando nossos objetos são representados por atributos categóricos, eles não possuem uma similaridade espacial. Para calcularmos a similaridade entre eles podemos primeiro transformar nosso vetor de atrbutos em um vetor binário indicando, para cada possível valor de cada atributo, se ele possui esse atributo ou não.

Com o vetor binário podemos utilizar a distância de Hamming definida por:

$$ H(x,y) = \sum_{i=1}^{n}{x_i != y_i} $$

Também é possível definir a distância de Jaccard como:

$$ J(x,y) = \frac{\sum_{i=1}^{n}{x_i == y_i} }{\sum_{i=1}^{n}{\max(x_i, y_i}) } $$





In [ ]:



    


# 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 [ ]:



    


# 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 [ ]:



    


# EXERCICIO
# Crie um RDD de chaves únicas utilizando flatMap
chavesRDD = (catPointsRDD
             .<COMPLETAR>
             .<COMPLETAR>
             .<COMPLETAR>
             )

chaves = dict((v,k) for k,v in chavesRDD.collect())
nchaves = len(chaves)
print (chaves, nchaves)



    











In [ ]:



    


assert chaves=={'alto': 2, 'caro': 0, 'baixo': 4, 'verde': 1, 'azul': 2, 'medio': 3, 'barato': 4, 'vermelho': 3}, 'valores incorretos!'
print ("OK")

assert nchaves==8, 'número de chaves incorreta'
print ("OK")



    











In [ ]:



    


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()



    











In [ ]:



    


# EXERCICIO
# Procure dentre os comandos do PySpark, um que consiga fazer o produto cartesiano da base com ela mesma
cartBinRDD = 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.<COMPLETAR>


# Aplique um mapa para calcular a Distância de Hamming e Jaccard entre os pares
hamRDD = cartBinParesRDD.<COMPLETAR>
jacRDD = cartBinParesRDD.<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 = 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 [ ]:



    


assert (Hmin.round(2), Hmax.round(2), Hmean.round(2)) == (0.00,5.00,2.40), 'valores incorretos'
print ("OK")
assert (Jmin.round(2), Jmax.round(2), Jmean.round(2)) == (0.60,4.00,1.90), 'valores incorretos'
print ("OK")

Content source: folivetti/BIGDATA

Similar notebooks:

notebook.community | gallery | about