Python 分布式计算框架 PP (Parallel Python):集群模式下的实践探索

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
大数据开发治理平台 DataWorks,不限时长
简介: 该文介绍了使用Parallel Python (PP) 在两台物理机上构建分布式计算集群的经验。PP是一个轻量级框架,旨在简化Python代码在多处理器系统和集群中的并行执行。文中通过设置子节点的IP、端口和密钥启动PP服务器,并在主节点创建PP实例进行负载均衡。实验使用官方的质数和计算示例,显示PP在集群模式下能有效利用多台机器的多核CPU,实现计算效率的显著提升。未来,作者计划进一步研究PP在更复杂任务和大规模集群中的应用潜力。

该文章是我在2016年发布在某sdn上的,搬运过来。


背景介绍

在现代计算领域,随着硬件性能的提升,特别是多核处理器和大规模集群的普及,分布式计算成为提高程序效率和处理大规模数据的关键手段。Parallel Python(简称PP)作为一款轻量级的分布式计算框架,旨在简化Python代码在SMP系统(多处理器或多核心)和集群环境中的并行执行。尽管网络上不乏关于PP单机多进程应用的教程,但在集群模式下PP的潜力却鲜少被深入挖掘。本文将分享在两台物理机(一台四核,一台双核)上搭建PP集群,实现分布式计算的实践经验。

平台配置与PP集群部署

部署PP集群的第一步是在所有参与计算的节点上安装Python环境,并通过pip install pp安装PP库。安装完成后,ppserver.py脚本将出现在Python的scripts目录中,它用于在子节点上启动监听端口。例如,在子节点上执行python ppserver.py -p 35000 -i 192.168.1.104 -s "123456",其中-p指定监听端口,-i是本地IP地址,-s后跟的是密钥,确保通信安全。

主节点则负责创建ppserver实例,需指定子节点的IP列表及相同的密钥,以建立安全的通信渠道。主节点通过动态负载均衡机制,将任务智能分配至各子节点,充分利用集群中每台机器的多核优势。

实战案例:求解质数和

为了验证PP在集群模式下的性能,我们采用了官方示例sum_primes.py,其主要功能是计算小于给定整数的所有质数之和。在本次实验中,我们将数据规模扩大,同时利用两台物理机的计算资源,观察PP的分布式计算能力。

Python

#!/usr/bin/python
# File: sum_primes.py
# Author: VItalii Vanovschi
# Desc: This program demonstrates parallel computations with pp module
# It calculates the sum of prime numbers below a given integer in parallel
# Parallel Python Software: http://www.parallelpython.com
 
import math, sys, time, datetime
import pp
 
def isprime(n):
    """Returns True if n is prime and False otherwise"""
    if not isinstance(n, int):
        raise TypeError("argument passed to is_prime is not of 'int' type")
    if n < 2:
        return False
    if n == 2:
        return True
    max = int(math.ceil(math.sqrt(n)))
    i = 2
    while i <= max:
        if n % i == 0:
            return False
        i += 1
    return True
 
def sum_primes(n):
    """Calculates sum of all primes below given integer n"""
    return sum([x for x in xrange(2,n) if isprime(x)])
 
print """Usage: python sum_primes.py [ncpus]
    [ncpus] - the number of workers to run in parallel, 
    if omitted it will be set to the number of processors in the system
"""
 
# tuple of all parallel python servers to connect with
#ppservers = ()
ppservers = ("192.168.1.104:35000",)
#ppservers=("*",)
 
if len(sys.argv) > 1:
    ncpus = int(sys.argv[1])
    # Creates jobserver with ncpus workers
    job_server = pp.Server(ncpus, ppservers=ppservers, secret="123456")
else:
    # Creates jobserver with automatically detected number of workers
    job_server = pp.Server(ppservers=ppservers, secret="123456")
 
print "Starting pp with", job_server.get_ncpus(), "workers"
 
# Submit a job of calulating sum_primes(100) for execution. 
# sum_primes - the function
# (100,) - tuple with arguments for sum_primes
# (isprime,) - tuple with functions on which function sum_primes depends
# ("math",) - tuple with module names which must be imported before sum_primes execution
# Execution starts as soon as one of the workers will become available
job1 = job_server.submit(sum_primes, (100,), (isprime,), ("math",))
 
# Retrieves the result calculated by job1
# The value of job1() is the same as sum_primes(100)
# If the job has not been finished yet, execution will wait here until result is available
result = job1()
 
print "Sum of primes below 100 is", result
 
start_time = time.time()
 
# The following submits 8 jobs and then retrieves the results
inputs = (500000, 500100, 500200, 500300, 500400, 500500, 500600, 500700, 500000, 500100, 500200, 500300, 500400, 500500, 500600, 500700)
#inputs = (1000000, 1000100, 1000200, 1000300, 1000400, 1000500, 1000600, 1000700)
jobs = [(input, job_server.submit(sum_primes,(input,), (isprime,), ("math",))) for input in inputs]
for input, job in jobs:
    print datetime.datetime.now()
    print "Sum of primes below", input, "is", job()
 
print "Time elapsed: ", time.time() - start_time, "s"
job_server.print_stats()


运行结果表明,PP成功地在两台机器上进行了负载均衡,总耗时显著减少,加速比达到了5.1倍,意味着相较于单机计算,PP集群模式有效利用了多台计算机的多核CPU资源,极大地提升了计算效率。


c:\Python27\python.exe test_pp_official.py
Usage: python sum_primes.py [ncpus]
    [ncpus] - the number of workers to run in parallel,
    if omitted it will be set to the number of processors in the system
Starting pp with 4 workers
Sum of primes below 100 is 1060
2016-08-28 19:07:26.579000
Sum of primes below 500000 is 9914236195
2016-08-28 19:07:33.032000
Sum of primes below 500100 is 9917236483
2016-08-28 19:07:33.035000
Sum of primes below 500200 is 9922237979
2016-08-28 19:07:33.296000
Sum of primes below 500300 is 9926740220
2016-08-28 19:07:33.552000
Sum of primes below 500400 is 9930743046
2016-08-28 19:07:33.821000
Sum of primes below 500500 is 9934746636
2016-08-28 19:07:34.061000
Sum of primes below 500600 is 9938250425
2016-08-28 19:07:37.199000
Sum of primes below 500700 is 9941254397
2016-08-28 19:07:37.202000
Sum of primes below 500000 is 9914236195
2016-08-28 19:07:41.640000
Sum of primes below 500100 is 9917236483
2016-08-28 19:07:41.742000
Sum of primes below 500200 is 9922237979
2016-08-28 19:07:41.746000
Sum of primes below 500300 is 9926740220
2016-08-28 19:07:41.749000
Sum of primes below 500400 is 9930743046
2016-08-28 19:07:41.752000
Sum of primes below 500500 is 9934746636
2016-08-28 19:07:41.756000
Sum of primes below 500600 is 9938250425
2016-08-28 19:07:43.846000
Sum of primes below 500700 is 9941254397
Time elapsed:  17.2770001888 s
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
         6 |         35.29 |      27.4460 |     4.574333 | 192.168.1.104:35000
        11 |         64.71 |      60.2950 |     5.481364 | local
Time elapsed since server creation 17.2849998474
0 active tasks, 4 cores


总结与展望

通过本次实战,我们不仅见证了PP在集群模式下的强大性能,也验证了其在分布式计算场景中的灵活性与高效性。PP不仅简化了并行计算的实现,还展示了其在跨平台、异构集群环境中的卓越适应能力。未来,我们期待进一步探索PP在更复杂计算任务和更大规模集群中的表现,挖掘其在科学计算、数据分析等领域的潜在价值。

参考文献

相关文章
|
6天前
|
关系型数据库 MySQL 数据库
如何使用Python的Flask框架来构建一个简单的Web应用
如何使用Python的Flask框架来构建一个简单的Web应用
12 0
|
5天前
|
人工智能 NoSQL 关系型数据库
主从模式(Master-Slave Architecture)在传统的计算机科学中指的是一种分布式系统架构
主从模式(Master-Slave Architecture)在传统的计算机科学中指的是一种分布式系统架构
22 5
|
4天前
|
缓存 前端开发 API
了解python中几个主流的网络框架
【6月更文挑战第21天】探索Python Web几个流行框架,了解各框架特性以适应不同场景需求。
24 1
|
7天前
|
存储 数据挖掘 索引
Python streamlit框架开发数据分析网站并免费部署
使用Python的Streamlit框架,开发了一个在线数据分析工具,替代Excel查看设备温度CSV数据。通过pandas读取数据,matplotlib绘制图表。程序处理CSV,提取所需列,计算最大最小平均值,用户可多选查看特定数据。[GitHub](https://github.com/yigedaigua/MGHB)上有完整代码,应用已部署至Streamlit Cloud。
|
3天前
|
Java 网络安全
分布式系统详解--框架(Zookeeper-简介和集群搭建)
分布式系统详解--框架(Zookeeper-简介和集群搭建)
10 0
|
3天前
|
分布式计算 资源调度 监控
分布式系统详解--框架(Hadoop-集群搭建)
分布式系统详解--框架(Hadoop-集群搭建)
15 0
|
11天前
|
存储 监控 负载均衡
Zookeeper 详解:分布式协调服务的核心概念与实践
Zookeeper 详解:分布式协调服务的核心概念与实践
14 0
|
11天前
|
机器人 测试技术 持续交付
Python进行自动化测试测试框架的选择与应用
【6月更文挑战第9天】本文介绍了Python自动化测试的重要性及选择测试框架的考量因素,如功能丰富性、易用性、灵活性和集成性。文中列举了常用的Python测试框架,包括unittest、pytest、nose2和Robot Framework,并提供了使用pytest进行单元测试的示例代码。此外,还展示了如何使用Robot Framework进行验收测试和Web UI测试。选择合适的测试框架对提升测试效率和软件质量至关重要,团队应根据项目需求、社区支持、集成性和学习曲线等因素进行选择。通过不断学习和实践,可以优化自动化测试流程,确保软件的稳定性和可靠性。
21 0
|
1月前
|
NoSQL Java 关系型数据库
【Redis系列笔记】分布式锁
分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。 分布式锁的核心思想就是让大家都使用同一把锁,只要大家使用的是同一把锁,那么我们就能锁住线程,不让线程进行,让程序串行执行,这就是分布式锁的核心思路
430 2
|
11天前
|
NoSQL 算法 Java
探讨redis分布式锁
探讨redis分布式锁
16 1