<jupyter_start><jupyter_text><jupyter_code># 1 - INSTALANDO O PYSPARK

!pip install pyspark
# 2 - IMPORTANDO O PYSPARK E A BIBLIOTECA DE FUNÇÕES DO PYSPARK

from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql.functions import col, udf
from pyspark.sql.types import DateType
from pyspark.sql.window import Window
# 3 - AUTENTICAÇÃO DA GCP

from google.colab import auth
auth.authenticate_user()
# 3 - SELECIONANDO O PROJETO - IMPORTANDO DIRETO DA BUCKET

project_id = 'projeto-final-grupo03'
!gcloud config set project projeto-final-grupo03
# 4 - FAZENDO DOWLOAD DO DOCUMENTO QUE SERÁ UTILIZADO - !gsutil cp gs://NomeDaBucket/NomeDoDocumento /tmp/NomeDoDocumento.csv

!gsutil cp gs://notebooks_pandas_gp03/pedagio_tratado.csv /tmp/pedagio_tratado.csv
#CONSULTAR PARA VER SE A TRANSFERÊNCIA FUNCIONOU

#!cat /tmp/pedagio_tratado.csv
# 5 - CONFIGURANDO A SPARKSESSION

spark = (SparkSession.builder
        .master("local")                   
        .appName("aprendendo-dataframes") 
        .config("spark.ui.port", "4050")
        .getOrCreate()) 
# CONSULTAR SE O SPARK ESTÁ INSTALADO

spark
# 6 - ESTRUTURA DO DATAFRAME

df = (spark
       .read
       .format("csv")
       .option("header", "true")
       .option("inferschema", "true")
       .option("delimiter", ",")
       .load("/tmp/pedagio_tratado.csv")
)
# 7 - VISUALIZANDO SCHEMA DO DATAFRAME E O TYPES DOS DADOS (STRING, INT, FLOAT)

df.printSchema()
#VISUALIZANDO O DATAFRAME

df.show()
# 8 - MUDAR O TIPO DE DATA PARA DATE COM  CAST

df = df.withColumn("Data", df.Data.cast("date"))
# 9 - VISUALIZANDO SCHEMA DO DATAFRAME PARA VER SE A MUDANÇA DA COLUNA "Data" DEU CERTO

df.printSchema()
df.show()
# 10 - RENOMEAR COLUNA COMO ID

df = df.withColumnRenamed('_c0', 'ID')
df.printSchema()
# 11 - FILTRO PARA IDENTIFICAR COLUNAS COM DADOS NULOS RELEVANTES OU NÃO

#TEM DADOS NULOS

df.filter(F.isnull("BR")).show()
df.filter(F.isnull("Volume_Total")).show()
df.filter(F.isnull("UF")).show()

#NÃO TEM DADOS NULOS

df.filter(F.isnull("ID")).show()
df.filter(F.isnull("Concessionaria")).show()
df.filter(F.isnull("Data")).show()
df.filter(F.isnull("Tipo_de_Veiculo")).show()
df.printSchema()
# 12 - VISUALIZANDO O QUE TEM NA COLUNA "Volume_Total"

df.select("UF").distinct().collect()<jupyter_output><empty_output><jupyter_text>FILTROS E CONSULTAS<jupyter_code>df.printSchema()
# 13 - FILTRO PARA IDENTIFICAR A QUANTIDADE ESPECIFICA DE PRAÇAS POR UF

df.groupBy(F.col("UF")).count().show()
# 14 - FILTRO PARA IDENTIFICAR A QUANTIDADE ESPECIFICA DE PRAÇAS POR "Concessionaria"

df.groupBy(F.col("Concessionaria")).count().show()
# 15 - FILTRO PARA IDENTIFICAR A QUANTIDADE ESPECIFICA DE PRAÇAS POR "Tipo_de_Veiculo"

df.groupBy(F.col("Tipo_de_Veiculo")).count().show()
# 16 - MÉDIA DE "Volume_Total" DE CARROS TRAFEGANDO EM TODAS AS VIAS

df.select(F.round(F.avg("Volume_Total"),2).alias('Média_de_Trafego')).show()

# MÉDIA DE CARROS POR ESTADO - grpdf = joined_df.groupBy(temp1.datestamp).max('diff').select(func.col("max(diff)").alias("maxDiff"))

df.groupBy(F.col("UF")).avg("Volume_Total").select(F.col("UF"), F.round(F.col("avg(Volume_Total)"), 2).alias('Média_de_Trafego_UF')).show()
df.printSchema()<jupyter_output>root
 |-- ID: integer (nullable = true)
 |-- Concessionaria: string (nullable = true)
 |-- Data: date (nullable = true)
 |-- Sentido: string (nullable = true)
 |-- Tipo_de_veiculo: string (nullable = true)
 |-- Volume_total: double (nullable = true)
 |-- BR: string (nullable = true)
 |-- UF: string (nullable = true)<jupyter_text>UTILIZANDO FUNÇÃO WINDOW<jupyter_code># 18 - UTILIZANDO FUNÇÃO PARA OTIMIZAR A PESQUISA DE QUAL O GRUPO DE CARROS MAIS PASSA PELAS RODOVIAS EM DETERMINADO ESTADO , ORGANIZANDO POR "UF" >> "Volume_Total" >> "BR"

w0 = Window.partitionBy(F.col("UF")).orderBy("Volume_Total", "UF", "BR")
df.withColumn("row_number", F.row_number().over(w0)).show()
# 19 - UTILIZANDO FUNÇÃO AGG PARA REALIZAR DOIS CALCULOS SOBRE O VOLUME DE CARROS PERCORRENDO AS RODOVIAS 

'''
A MÉDIA DAS RENDAS DOS CLIENTES
A SOMA TOTAL DAS RENDAS DOS CLIENTES
'''

(df.withColumn("row", F.row_number().over(w0))
 .withColumn("avg", F.avg(F.col("Volume_Total")).over(w0))
 .withColumn("sum", F.sum(F.col("Volume_Total")).over(w0))
  .where(F.col("row") == 1).select("row","UF", "BR", "avg", "sum").show()
)
df.printSchema()
#20 - SALVANDO O ARQUIVO NORMALIZADO NO BUCKET

(df.write.format("csv").option("header", "true")
                        .option("inferschema", "true")
                        .option("delimiter", ",")
                        .save("pedagio_pyspark"))
!gsutil cp -r pedagio_pyspark gs://notebooks_pyspark_sql_gp03<jupyter_output>Copying file://pedagio_pyspark/._SUCCESS.crc [Content-Type=application/octet-stream]...
Copying file://pedagio_pyspark/.part-00000-732acd99-8e9c-476c-8f83-64ec9ad8fc34-c000.csv.crc [Content-Type=application/octet-stream]...
Copying file://pedagio_pyspark/part-00000-732acd99-8e9c-476c-8f83-64ec9ad8fc34-c000.csv [Content-Type=text/csv]...
Copying file://pedagio_pyspark/_SUCCESS [Content-Type=application/octet-stream]...
\ [4 files][  4.0 MiB/  4.0 MiB]                                                
Operation completed over 4 objects/4.0 MiB.
StarcoderdataJupyterStructuredCleanDedup