并行计算框架Polars、Dask的数据处理性能对比( 二 ) _框架

2、Dask函数功能与上面一样，所以我们把代码整合在一起：
import dask.dataframe as dd from dask.distributed import Client import timedef extraction():path1 = "yellow_tripdata/yellow_tripdata_2014-01.parquet"df_trips = dd.read_parquet(path1)path2 = "taxi+_zone_lookup.parquet"df_zone = dd.read_parquet(path2)return df_trips, df_zonedef transformation(df_trips, df_zone):df_trips = mean_test_speed_dask(df_trips)df = df_trips.merge(df_zone, how="inner", left_on="PULocationID", right_on="LocationID")df = df[["Borough", "Zone", "trip_distance"]]df = get_Queens_test_speed_dask(df)df = round_column(df, "trip_distance", 2)df = rename_column(df, "trip_distance", "mean_trip_distance")df = sort_by_columns_desc(df, "mean_trip_distance")return dfdef loading_into_parquet(df_dask):df_dask.to_parquet("yellow_tripdata_dask.parquet", engine="fastparquet")def mean_test_speed_dask(df_dask):df_dask = df_dask.groupby("PULocationID").agg({"trip_distance": "mean"})return df_daskdef get_Queens_test_speed_dask(df_dask):df_dask = df_dask[df_dask["Borough"] == "Queens"]return df_daskdef round_column(df, column, to_round):df[column] = df[column].round(to_round)return dfdef rename_column(df, column_old, column_new):df = df.rename(columns={column_old: column_new})return dfdef sort_by_columns_desc(df, column):df = df.sort_values(column, ascending=False)return dfdef main():print("Starting ETL for Dask")start_time = time.perf_counter()client = Client() # Start Dask Clientdf_trips, df_zone = extraction()end_extract = time.perf_counter()time_extract = end_extract - start_timeprint(f"Extraction Parquet end in {round(time_extract, 5)} seconds")print("Transforming...")df = transformation(df_trips, df_zone)end_transform = time.perf_counter()time_transformation = time.perf_counter() - end_extractprint(f"Transformation end in {round(time_transformation, 5)} seconds")print("Loading...")loading_into_parquet(df)load_transformation = time.perf_counter() - end_transformprint(f"Loading end in {round(load_transformation, 5)} seconds")print(f"End ETL for Dask in {str(time.perf_counter() - start_time)}")client.close() # Close Dask Clientif __name__ == "__main__":main()测试结果对比1、小数据集我们使用164 Mb的数据集，这样大小的数据集对我们来说比较小，在日常中也时非常常见的。
下面是每个库运行五次的结果:
Polars