<jupyter_start><jupyter_text>Data Science Academy Big Data Real-Time Analytics com Python e Spark Capítulo 8  *********** Atenção: *********** Utilize Java JDK 1.8 e Apache Spark 2.4.2 Java JDK 1.8:    https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html *Caso receba mensagem de erro "name 'sc' is not defined", interrompa o pyspark e apague o diretório metastore_db no mesmo diretório onde está este Jupyter notebook* Acesse http://localhost:4040 sempre que quiser acompanhar a execução dos jobs  Agrupamento com Pares RDD (Pair RDD) Tipo especial de RDD que armazena pares chave-valor. Útil quando é necessáirio armazenar dados que possuem uma chave e diversos valores (por exemplo, todas as transações de um cliente, geradas em tempo real).  mapValues() countByKey() groupByKey() reduceByKey() aggregateByKey()<jupyter_code># Importando arquivo csv e criando um RDD
carros = sc.textFile("data/carros.csv")
carros.take(20)
# Criando uma Pair RDD
carrosPairRDD = carros.map(lambda x: (x.split(",")[0], x.split(",")[7]))
carrosPairRDD.take(20)
# Removendo o cabeçalho
header = carrosPairRDD.first()
carrosPairRDD2 = carrosPairRDD.filter(lambda line: line != header)
# Encontra o valor de HP por marca de carro e adiciona 1 a cada registro "Make/HP"
carrosPairRDD3 = carrosPairRDD2.mapValues(lambda x: (x, 1))
carrosPairRDD3.collect()
# Aplica redução por key (reduceByKey)
# E calcula o total de HP por fabricante e o total de automóveis por fabricante
fabricantes =  carrosPairRDD3.reduceByKey(lambda x, y: (int(x[0]) + int(y[0]), x[1] + y[1])) 
fabricantes.collect()
# Calculando a média de HP dividindo pela contagem total
fabricantes.mapValues(lambda x: int(x[0])/int(x[1])).collect()<jupyter_output><empty_output><jupyter_text>Acumuladores e Broadcast O Spark faz uma cópia do código que você escreveu para processar os dados e executa essas cópias, uma por node do cluster. Qualquer variável criada no código é local ao node. O Spark gera cópias dessas variáveis locais, uma em cada node, que agem de forma independente. Mas e se precisamos que a mesma variável seja manipulada de forma única através de todo o cluster? Usamos Acumuladores e Broadcast. Variável Broadcast - read-only, é compartilhada em todo o cluster. Variável Accumulator - é compartilhada em todo o cluster, mas pode ser atualizada em cada node do cluster.<jupyter_code># Inicializando variáveis Accumulator
sedanCount = sc.accumulator(0)
hatchbackCount = sc.accumulator(0)
# Inicializando variáveis Broadcast
sedanText = sc.broadcast("sedan")
hatchbackText = sc.broadcast("hatchback")
def splitLines(line) :

    global sedanCount
    global hatchbackCount

    # Usa a variável Broadcast para comparar e adiciona a contagem ao accumulator
    if sedanText.value in line:
        sedanCount +=1
    if hatchbackText.value in line:
        hatchbackCount +=1
        
    return line.split(",")
# Map()
splitData = carros.map(splitLines)
# Ação para executar a transformação (lazy evaluation)
splitData.count()
print(sedanCount, hatchbackCount)<jupyter_output>92 67<jupyter_text>Partições Sempre que criamos RDD's, esses objetos são divididos em partições e essas partições são distribuídas através dos nodes do cluster. Por default, os RDD's são sempre particionados. Essas partições precisam ser configuradas quando se trabalha com grandes clusters.<jupyter_code>fabricantes.getNumPartitions()
# Especificando o Número de Partições
collData = sc.parallelize([3,5,4,7,4], 3)
collData.cache()
collData.count()
collData.getNumPartitions()
print (sc.defaultParallelism)<jupyter_output>8
{'fraction_non_alphanumeric': 0.08711383245198998, 'fraction_numerical': 0.015029223490119677, 'mean_word_length': 5.361061946902655, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '374604', 'license': 'MIT', 'n_tokens_mistral': 1249, 'n_tokens_neox': 1205, 'n_words': 408, 'path': 'Apache Spark/Spark-SQL/02-Cap08-Spark-Pair-RDD-Acumuladores-Broadcast.ipynb', 'repo_name': 'marxcerqueira/Spark-Real-Time-Analytics', 'size': '3593'}