Python库巡礼--Dask

import pandas as pdimport dask.dataframe as ddimport time# Pandas（内存不足会崩溃）start = time.time()df_pd = pd.read_csv("large_file.csv")  # 10GB文件可能OOMprint(f"Pandas加载时间: {time.time()-start:.2f}s")# Dask（分块并行处理）start = time.time()df_dd = dd.read_csv("large_file.csv", blocksize="64MB")  # 自动分块print(f"Dask构建时间: {time.time()-start:.2f}s (懒加载，实际未读取)")# 执行计算（触发实际计算）result = df_dd.groupby("category")["sales"].sum().compute()  # .compute()触发执行print(result)

import dask.array as daimport numpy as np# 创建10亿元素的虚拟数组（不占内存）x = da.random.random((10000, 100000), chunks=(1000, 10000))  # chunks定义分块大小# 构建计算图（懒加载）y = x + x.T.mean(axis=0)z = da.fft.fft(y)  # 快速傅里叶变换# 可视化任务图（调试用）z.visualize(filename="task_graph.png")  # 生成PNG展示计算依赖# 执行计算（自动并行）result = z.compute()print(result.shape)

from dask.distributed import Client, progressimport dask.dataframe as dd# 启动本地集群（4个工作进程）client = Client(n_workers=4, threads_per_worker=2)print(client.dashboard_link) # 读取数据df = dd.read_parquet("s3://my-bucket/data/*.parquet")  # 支持S3/HDFS# 复杂ETL流程result = (    df[df.amount > 100]    .groupby("user_id")    .agg({"amount": "sum", "timestamp": "max"})    .compute()  # 触发分布式计算)# 进度条future = client.compute(result)progress(future)  # 实时显示进度client.close()  # 关闭集群