Python学习（1）：使用Python的Dask库实现并行计算
目录
一、Dask介绍
二、使用说明
安装
三、测试
1、单个文件中实现功能
2、运行多个可执行文件
最近在写并行计算相关部分，用到了python的Dask库。
Dask官网：Dask | Scale the Python tools you love
一、Dask介绍
Dask是一个灵活的并行和分布式计算库，旨在处理大规模数据集。它提供了类似于Pandas 和 NumPy 的数据结构，但能够有效处理比内存更大的数据集。通过使用Dask，可以在单台机器或分布式集群中运行，更方便处理大规模数据。
Dask是一个用于Python的并行计算模块，从单机多核扩展到拥有数千台机器的数据中心。它既由低级任务API，也有更高级面向数据的API。低级任务API支持Dask与多种Python库的集成，公共API为围绕Dask发展的各种工具的生态系统提供了基础。
Dask相较于Spark这些大数据处理框架，更轻量级。Dask更侧重与其他框架，如：Numpy、Pandas，Scikit-learning相结合，从而使其能更加方便进行分布式并行计算。
Dask存在三种最基本的数据结构，分别是：Arrays、Dataframes以及Bags。
二、使用说明
安装
pip install dask
python -m pip install "dask[array]"
python -m pip install "dask[distributed]"
python -m pip install "dask[dataframe]"先测试是否已经安装了模块，命令行进入到python3编辑器：
from dask.distributed import Client, progress
没有报缺少模块错误，则说明是可以正常执行的。
三、测试
1、单个文件中实现功能
下述的主要数据处理在定义计算任务函数calculate_value(num)中，即将计算任务函数处理32次。
from dask.distributed import Client, progress
import time# 定义计算任务的函数
def calculate_value(num):num_float = float(num) * 0.33num_double = float(num) * 0.33 return num_float, num_double# 设置Dask客户端
def setup_client():from dask.distributed import Client, LocalClustercluster = LocalCluster()client = Client(cluster)scheduler_info = client.scheduler_info()ncores = sum(worker['nthreads'] for worker in scheduler_info['workers'].values())print(f"Connected to Dask cluster with {ncores} cores")return client# 提交任务并收集结果
def submit_tasks(client, num, num_tasks=32):# 创建任务列表tasks = [client.submit(calculate_value, num) for _ in range(num_tasks)]# 等待所有任务完成，并显示进度progress(tasks)# 收集结果results = [task.result() for task in tasks]return results# 主函数
def main():num = 558558571 # 这是您要处理的数字client = setup_client() # 设置Dask客户端# 提交32个任务results = submit_tasks(client, num)# 打印结果for i, (num_float, num_double) in enumerate(results):print(f"Task {i+1} - num_float: {num_float}, num_double: {num_double}")# 关闭客户端连接client.close()if __name__ == "__main__":main()运行上述的python程序：
python3 my_dask_script.py执行结果如下：
此时表示运行了32个task。
在运行的时候如果提示：
表明 dask-scheduler 无法启动，原因是端口 8787 已经被占用了。
解决方法：
1、查找并终止占用端口 8787 的进程
（1）先安装lsof：
apt install lsof（2）查看占用端口进程：
lsof -i :8787
（3）通过进程的 PID 使用 kill 命令终止该进程：
kill -9 PID
2、修改 dask-scheduler 使用的端口
dask-scheduler --port 8888
再次重新启动查看 dask-scheduler 使用的端口：
dask-scheduler
2、运行多个可执行文件
我在同目录中创建了一个test.cc文件，为简单的打印数据，内容如下：
#include <iostream>
#include <iomanip>int main() {int num = 558558571;float num_float = static_cast<float>(num) * 0.33;double num_double = static_cast<double>(num) * 0.33;std::cout << "num value: " << num << std::endl;std::cout << std::fixed << std::setprecision(2);std::cout << "num_float value: " << num_float << std::endl;std::cout << "num_double value: " << num_double << std::endl;return 0;
}此时将上述的test.cc编译：
g++ -o main test.cc
然后新建一个my_dask_script.py文件，内容如下：
from dask.distributed import Client, LocalCluster
import os# 定义执行外部程序的函数
def run_external_program():cmd = './main' # 您的外部程序命令os.system(cmd) # 使用os.system来执行命令# 设置Dask客户端
def setup_client():from dask.distributed import Client, LocalClustercluster = LocalCluster()client = Client(cluster)scheduler_info = client.scheduler_info()ncores = sum(worker['nthreads'] for worker in scheduler_info['workers'].values())print(f"Connected to Dask cluster with {ncores} cores")return client# 提交任务到Dask集群
def submit_tasks(client, num_tasks=32):futures = [client.submit(run_external_program) for _ in range(num_tasks)]return futures# 主函数
def main():client = setup_client() # 设置Dask客户端futures = submit_tasks(client) # 提交任务# 等待所有任务完成client.gather(futures)# 关闭客户端连接client.close()if __name__ == "__main__":main()运行结果：
此时表示上述的可执行文件main已运行了32份。