diff --git a/One-Billion-Row-Challenge.png b/One-Billion-Row-Challenge.png new file mode 100644 index 0000000..eb5c635 Binary files /dev/null and b/One-Billion-Row-Challenge.png differ diff --git a/README.md b/README.md index 306acaa..a9cad96 100644 --- a/README.md +++ b/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 diff --git a/src/create_measurements.py b/src/create_measurements.py index b6db21f..48891f5 100644 --- a/src/create_measurements.py +++ b/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)) diff --git a/src/using_pandas.py b/src/using_pandas.py index 05f88e8..8256062 100644 --- a/src/using_pandas.py +++ b/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 diff --git a/src/using_polars.py b/src/using_polars.py index b6cdd44..e6f9eb4 100644 --- a/src/using_polars.py +++ b/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(), diff --git a/src/using_pyspark.py b/src/using_pyspark.py new file mode 100644 index 0000000..184211a --- /dev/null +++ b/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.") diff --git a/src/using_python.py b/src/using_python.py index f945256..1610e53 100644 --- a/src/using_python.py +++ b/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.") \ No newline at end of file diff --git a/src/using_python_old.py b/src/using_python_old.py index 76c59a8..eb6cb6d 100644 --- a/src/using_python_old.py +++ b/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.")