Dask 高效的Python并行计算库

CxdPvE

官网【https://dask.org/】

官方文档 【https://docs.dask.org/】

没有学的很精通，也无法从小到大，由浅到深的讲解，所以记录一些使用场景吧，这种记录方式也挺实用的。

场景1. 读取较大的数据集

Dask有着丰富的读取数据的接口，而且接口与常用的 pandas numpy 数据接口基本相同，非常容易上手。
由于专业关系，我最常用的数据格式为 netCDF ,通常需要借助 xarray 库进行读取保存等操作。但非常好的消息是 xarray 与 Dask 建立了十分友好的互帮互助，共同进步关系， xarray 有着非常方便的参数接口，不需要太多额外配置就能够直接使用 Dask .

  1. # hycom 在线数据，TB级别，传入chunks分块参数，直接使用Dask能够秒级别导入
  2. data_global = xr.open_dataset('http://tds.hycom.org/thredds/dodsC/GLBy0.08/expt_93.0',decode_times=False,chunks={"time":100})
  4. # 如果是多个小文件，使用mfdataset，传入parallel并行参数，秒级别打开多个数据，并且能够按照某个维度整合
  5. dataall = xr.open_mfdataset(url_exist,parallel = True,decode_times=False,combine='nested',concat_dim="time")
  7. # 注意该函数对本地文件夹循环也非常好用
  8. dataall = xr.open_mfdataset('data_all/*.nc',parallel=True)

场景2. 保存和下载数据

使用 save_mfdataset(filelist,paths) 并行保存数据。注意前提是数据为 dask 类型，否则相当于循环调用 to_netcdf() 。

  1. import xarray as xr
  3. data_global = xr.open_dataset('http://tds.hycom.org/thredds/dodsC/GLBy0.08/expt_93.0',decode_times=False,chunks={"time":100})
  5. data_latest = data_global.sel(lat = slice(2,42),lon = slice(104,132),depth=slice(0,1001))
  7. from datetime import datetime,timedelta
  8. import pandas as pd
  9. import numpy as np
  10. def return_latest_time():
  11.     date_start = '2000-01-01 00:00:00'
  12.     date_list = []
  13.     for i in data_latest.time.data:
  14.         date_list.append(pd.to_datetime(date_start)+timedelta(hours = i))
  15.     return date_list
  16. date_list = return_latest_time()
  18. data_latest['time'] = pd.to_datetime(date_list) # 重新更新文件时间
  20. date_time = pd.to_datetime(date_list)
  22. data_latest_now =[]
  23. data_latest_path= []
  24. for date in date_time:
  25.     if (date.hour == 0 or date.hour == 12) and date.year>2019:
  26.         data_latest_now.append(data_latest.sel(time = date))
  27.         data_latest_path.append("/data/hycom_2020_latest/{}.nc".format(str(date)))
  28.         print(date,"已完成！！！")         
  30. xr.save_mfdataset(data_latest_now,data_latest) # 并行保存
  32. #### 下载  一天两次 普通保存
  34. # for date in date_time:
  35. #     if date.hour == 0 or date.hour == 12 :
  36. #         data_latest_now = data_latest.sel(time = date)
  37. #         data_latest_now.to_netcdf("/data/hycom_2018_latest/{}.nc".format(str(date)))
  38. #         print(date,"已完成！！！")

场景3. Delay

delay 是先将运算搁置，等到 compute 的时候再进行计算，非常方便调试大数据程序。

场景3. Persist

persist 作用是将dask先进行一次运算，结果任然保持dask格式，后面的运算仍支持dask并行计算。