Adicionando arquivo using_pyspark.py

README.md

@@ -59,16 +59,19 @@ Para executar os scripts deste projeto, você precisará das seguintes bibliotec
 
 ## Resultados
 
-Os testes foram realizados em um laptop equipado com um processador M1 da Apple e 8GB de RAM. As implementações utilizaram abordagens puramente Python, Pandas, Dask, Polars e DuckDB. Os resultados de tempo de execução para processar o arquivo de 1 bilhão de linhas são apresentados abaixo:
+Os testes foram realizados em um notebook equipado com um processador AMD® Ryzen 7 5700u with radeon graphics × 16 e 32GB de RAM. As implementações utilizaram abordagens puramente Python, Pandas, Dask, Polars, Pyspark e DuckDB. Os resultados de tempo de execução para processar o arquivo de ~~1 bilhão~~ 100 milhões de linhas são apresentados abaixo:
 
 | Implementação | Tempo |
 | --- | --- |
-| Bash + awk | 25 minutos |
-| Python | 20 minutos |
-| Python + Pandas | 263 sec |
-| Python + Dask | 155.62 sec  |
-| Python + Polars | 33.86 sec |
-| Python + Duckdb | 14.98 sec |
+| Python | 1 min e 68 sec |
+| Python + Pandas | 41.74 sec |
+| Python + Dask | 38.23 sec  |
+| Python + Spark | 19.68 sec  |
+| Python + Duckdb | 2.64 sec |
+| Python + Polars | 2.08 sec |
+
+![alt text](One-Billion-Row-Challenge.png)
+
 
 Obrigado por [Koen Vossen](https://github.com/koenvo) pela implementação em Polars e [Arthur Julião](https://github.com/ArthurJ) pela implementação em Python e Bash 
 

src/create_measurements.py

@@ -123,7 +123,7 @@ def main():
     """
     main program function
     """
-    num_rows_to_create = 1000000
+    num_rows_to_create = 100000000
     weather_station_names = []
     weather_station_names = build_weather_station_name_list()
     print(estimate_file_size(weather_station_names, num_rows_to_create))

src/using_pandas.py

@@ -4,7 +4,7 @@
 
 CONCURRENCY = cpu_count()
 
-total_linhas = 1_000_000_000  # Total de linhas conhecido
+total_linhas = 100_000_000  # Total de linhas conhecido
 chunksize = 100_000_000  # Define o tamanho do chunk
 filename = "data/measurements.txt"  # Certifique-se de que este é o caminho correto para o arquivo
 

src/using_polars.py

@@ -9,7 +9,7 @@ def create_polars_df():
     return (
 
         pl.scan_csv("data/measurements.txt", separator=";", has_header=False, new_columns=["station", "measure"], schema={"station": pl.String, "measure": pl.Float64})
-        .group_by(by="station")
+        .group_by("station")
         .agg(
             max = pl.col("measure").max(),
             min = pl.col("measure").min(),

src/using_pyspark.py

@@ -0,0 +1,48 @@
+from pyspark.sql import SparkSession
+from pyspark.sql import functions as F
+import time
+
+# Setei esses parâmetros para minha máquina
+# Linux 11, 32Gb de RAM, processador AMD® Ryzen 7 5700u with radeon graphics × 16 
+spark = SparkSession.builder \
+    .appName("OneBillionRowChallenge") \
+    .master("local[12]") \
+    .config("spark.sql.shuffle.partitions", "16") \
+    .config("spark.driver.memory", "24g") \
+    .config("spark.executor.memory", "24g") \
+    .config("spark.driver.maxResultSize", "8g") \
+    .getOrCreate()
+
+def processar_temperaturas(file_path):
+    df = spark.read.csv(file_path, sep=';', header=False, inferSchema=True) \
+    .toDF("station", "measure")
+
+    df = df.repartition(16)
+
+    df = df.groupBy("station") \
+        .agg(
+            F.min("measure").alias("min"),
+            F.max("measure").alias("max"),
+            F.mean("measure").alias("mean")
+        ) \
+        .orderBy("station") ##.show(5) # Vou exibir apenas os 5 primeiros registros
+
+    return df
+
+
+if __name__ == "__main__":
+    file_path = "data/measurements.txt" 
+
+    print("Iniciando o processamento do arquivo.")
+    start_time = time.time()  # Tempo de início
+
+    try:
+        resultados = processar_temperaturas(file_path)
+
+        end_time = time.time() # Tempo de término
+
+    finally:
+        spark.stop()
+
+    print(resultados)
+    print(f"\nProcessamento concluído em {end_time - start_time:.2f} segundos.")

src/using_python.py

@@ -3,7 +3,7 @@
 from tqdm import tqdm  # barra de progresso
 import time
 
-NUMERO_DE_LINHAS = 1_000_000_000
+NUMERO_DE_LINHAS = 100_000_000
 
 def processar_temperaturas(path_do_csv):
     # utilizando infinito positivo e negativo para comparar
@@ -51,7 +51,11 @@ def processar_temperaturas(path_do_csv):
 
     end_time = time.time()  # Tempo de término
 
-    for station, metrics in resultados.items():
+    # Exibindo apenas os 5 primeiros resultados
+    for i, (station, metrics) in enumerate(resultados.items()):
+        if i == 5:
+            break
         print(station, metrics, sep=': ')
+    print("...")
 
     print(f"\nProcessamento concluído em {end_time - start_time:.2f} segundos.")

src/using_python_old.py

@@ -1,8 +1,8 @@
 from csv import reader
 from collections import defaultdict
+from pathlib import Path
 import time
 
-from pathlib import Path
 
 def processar_temperaturas(path_do_txt: Path):
     print("Iniciando o processamento do arquivo.")
@@ -57,6 +57,14 @@ def processar_temperaturas(path_do_txt: Path):
     # Formatando os resultados para exibição
     formatted_results = {station: f"{min_temp:.1f}/{mean_temp:.1f}/{max_temp:.1f}" for station, (min_temp, mean_temp, max_temp) in sorted_results.items()}
 
+    # Exibindo apenas os 5 primeiros resultados
+    print("\nTabela de resultados (primeiros 5 registros):")
+    for i, (station, stats) in enumerate(formatted_results.items()):
+        if i == 5:
+            break
+        print(f"{station}: {stats}")
+    print("...")
+
     end_time = time.time()  # Tempo de término
     print(f"Processamento concluído em {end_time - start_time:.2f} segundos.")