<jupyter_start><jupyter_text>SPARK_COVID - EDUARDO CRUZ  Projeto para integrar dados públicos da COVID e gerar análises.<jupyter_code># Importar dependencias:
from pyspark.sql.functions import *
from pyspark.sql.types import *


# Setando configurações para permitir particionamento dinamico.
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
spark.conf.set("spark.hadoop.hive.exec.max.dynamic.partitions", "30000")


# Criando schema
columnsList = [ StructField("regiao", StringType()), 
               StructField("estado", StringType()), 
               StructField("municipio", StringType()), 
               StructField("coduf", StringType()), 
               StructField("codmun", StringType()), 
               StructField("codRegiaoSaude", StringType()), 
               StructField("nomeRegiaoSaude", StringType()), 
               StructField("data", DateType()), 
               StructField("semanaEpi", StringType()), 
               StructField("populacaoTCU2019", LongType()), 
               StructField("casosAcumulado", LongType()), 
               StructField("casosNovos", LongType()), 
               StructField("obitosAcumulado", LongType()), 
               StructField("obitosNovos", LongType()), 
               StructField("Recuperadosnovos", LongType()), 
               StructField("emAcompanhamentoNovos", LongType()), 
               StructField("interiormetropolitana", StringType())]

covidSchema = StructType(columnsList)

# Fazendo a leitura do CSV e vinculando o schema criado.
covidDF = spark.read.option("header","true").option("delimiter", ";").schema(covidSchema).csv("/user/eduardo/spark_covid/tz/covid/*.csv")
# Salvar dados em uma tabela hive particionada por "municipio"
# Rotina executada com sucesso no spark-shell, porem aqui no pyspark não foi possível alterar o limite de partições, mesmo
# após várias tentativas.
covidDF.write.mode("overwrite").partitionBy("municipio").format("hive").saveAsTable("covid")

# Gerando a visualização do Total de:
# - Casos Recuperados 
# - Em Acompanhamento
casosRecAcom = covidDF.sort(desc("data")).filter(col("regiao") == "Brasil").limit(1).agg(format_number(max("Recuperadosnovos"), 1).alias("Casos_Recuperados"),format_number(max("emAcompanhamentoNovos"),1).alias("Em_Acompanhamento"))
casosRecAcom.show()
# Gerando a visualização do Total de:
# - Casos Confirmados
#  - Acumulado
#  - Casos novos 
#  - Incidência 

casosConfirmados = covidDF.sort(desc("data")).filter(col("regiao") == "Brasil").limit(1).agg(format_number(max("casosAcumulado"),1).alias("Casos_Acumulados"),format_number(max("casosNovos"),1).alias("Casos_Novos"), format_number(sum((col("casosAcumulado") / col("populacaoTCU2019")) * 100000),1).alias("Incidencia"))
casosConfirmados.show()

# Gerando a visualização do Total de:
# - Óbitos Confirmados
#   - Acumulado
#   - Casos Novos
#   - Letalidade
#   - Mortalidade 

obitosConfirmados = covidDF.sort(desc("data")).filter(col("regiao") == "Brasil").limit(1).agg(format_number(max("obitosAcumulado"),1).alias("Obitos_Acumulados"),format_number(max("obitosNovos"),1).alias("Obitos_Novos"), format_number(sum((col("obitosAcumulado") / col("casosAcumulado")) * 100),1).alias("Letalidade"), format_number(sum((col("obitosAcumulado") / col("populacaoTCU2019")) * 100000),1).alias("Mortalidade"))
obitosConfirmados.show()

# Salvando a visualização de Casos Recuperados e Casos em acompanhamento em uma tabela Hive
casosRecAcom.write.mode("overwrite").format("hive").saveAsTable("casosRecAcom1")
!hdfs dfs -ls /user/hive/warehouse/casosrecacom1/
# Salvando a visualização de Casos Confirmados em formato parquet e compressão snappy
casosConfirmados.write.mode("overwrite").option("compression", "snappy").parquet("/user/eduardo/casosConfirmados")
!hdfs dfs -ls /user/eduardo/casosConfirmados/
# Salvando a visualização de Obitos Confirmados em um topico no KAFKA

obitosConfirmados.selectExpr("to_json(struct(*)) AS value").write.format("kafka").option("kafka.bootstrap.servers", "kafka:9092").option("topic", "topic-obitosConfirmados").save()
# Lendo os dados gravados no Kakfa e exibindo os valores

kafka_df = spark \
.read \
.format("kafka") \
.option("kafka.bootstrap.servers", "kafka:9092") \
.option("subscribe", "topic-obitosConfirmados") \
.option("startingOffsets", "earliest") \
.load()

topic_string = kafka_df.select(col("value").cast("string"))
topic_string.show()

# Criar visualização da Sintese dos dados de casos, obitos, incidencia e mortalidade.
# - Nessa primeira etapa foi feito o agrupamento por estado e região. 
sinteseEstados = covidDF.filter(col("regiao").isNotNull()).groupBy("estado", "regiao").agg(max("casosAcumulado").alias("casosAcumulado"), max("obitosAcumulado").alias("obitosAcumulado"), max("populacaoTCU2019").alias("populacaoTCU2019"), last("data").alias("data")).sort(asc("regiao"),asc("estado"),desc("data"))
sinteseEstados.show()
# Nessa etapa foi feito o resumo por Região
sinteseRegiao = sinteseEstados.groupBy("regiao").agg(sum("casosAcumulado").alias("Casos"), sum("obitosAcumulado").alias("Obitos"), sum("populacaoTCU2019").alias("populacaoRegiao"), last("data").alias("Atualizacao")).sort(asc("regiao"))
sinteseRegiao.show()
# Nessa é feito o resultado geral, incluindo os cálculos de incidencia e mortalidade.
sinteseGeral = sinteseRegiao.withColumn("Incidencia", (col("Casos") / col("populacaoRegiao")) * 100000).withColumn("Mortalidade", (col("Obitos") / col("populacaoRegiao")) * 100000)
sinteseGeral.show()
# Para finalizar, nessa etapa foi feita a formatação dos valores para uma melhor leitura. 
sinteseFinal = sinteseGeral.groupBy("regiao").agg(format_number(last("Casos"),1).alias("Casos"),format_number(last("Obitos"),1).alias("Obitos"),format_number(last("Incidencia"),1).alias("Incidencia"),format_number(last("Mortalidade"),1).alias("Mortalidade"),last("Atualizacao").alias("Atualizacao"))
sinteseFinal.show()<jupyter_output>+------------+------------+---------+----------+-----------+-----------+
|      regiao|       Casos|   Obitos|Incidencia|Mortalidade|Atualizacao|
+------------+------------+---------+----------+-----------+-----------+
|      Brasil|18,855,015.0|526,892.0|   8,972.3|      250.7| 2021-07-06|
|Centro-Oeste| 1,916,619.0| 49,207.0|  11,760.5|      301.9| 2021-07-06|
|    Nordeste| 4,455,737.0|107,824.0|   7,807.3|      188.9| 2021-07-06|
|       Norte| 1,732,862.0| 43,845.0|   9,401.9|      237.9| 2021-07-06|
|     Sudeste| 7,138,803.0|245,311.0|   8,078.2|      277.6| 2021-07-06|
|         Sul| 3,611,041.0| 80,705.0|  12,046.4|      269.2| 2021-07-06|
+------------+------------+---------+----------+-----------+-----------+
{'fraction_non_alphanumeric': 0.19012715033657443, 'fraction_numerical': 0.04098728496634256, 'mean_word_length': 4.471358428805237, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '442087', 'license': 'MIT', 'n_tokens_mistral': 2513, 'n_tokens_neox': 2294, 'n_words': 369, 'path': 'pyspark/spark_covid.ipynb', 'repo_name': 'eduardocruzs/spark_covid', 'size': '6685'}