Skip to content

Commit 96631ed

Browse files
author
wangchao
committed
更新
1 parent 2659027 commit 96631ed

8 files changed

+211
-113
lines changed

‎.gitignore‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,4 +128,5 @@ dmypy.json
128128
# Pyre type checker
129129
.pyre/
130130

131-
temp.py
131+
temp.py
132+
.DS_Store

‎docs/chapter6/discovering_pp.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ PP模块有两种方式来执行并行代码. 第一种方式基于SMP架构,即
1616
在PP的API中有一个名为Server的类,使用该类可以实现在本地和远程的进程间封装和分派任务. Server的构造函数(\__init\__)中有几个参数比较重要:
1717

1818
-**ncpus**: 该参数用于指定执行任务的工作进程数量. 若没有指定该参数,则会自动根据机器上处理器/核心的数量来创建工作进程的总数,以优化资源的使用。
19-
-**ppservers**: 该参数是一个元组,该元组的元素为并行Python执行服务器(PPES)的名称或IP地址. 一个PPES由连入网络的机器组成. 且该机器通过`ppsever.py`共组运行并等待待执行的任务. 其他参数的说明请参阅<http://www.parallelpython.com/content/view/15/30/>
19+
-**ppservers**: 该参数是一个元组,该元组的元素为**并行Python执行服务器**(Parallel Python Execution Servers - PPES)的名称或IP地址. 一个PPES由连入网络的机器组成. 且该机器通过`ppsever.py`共组运行并等待待执行的任务. 其他参数的说明请参阅<http://www.parallelpython.com/content/view/15/30/>
2020

2121
`Server`类的实例拥有很多方法,其中`submit`方法允许我们分配任务到各个工作进程. `submit`函数具有如下签名:
2222

docs/chapter6/在SMP架构上使用pp模块计算斐波那契序列.md renamed to docs/chapter6/using_pp_to_calculate_the_fibonacci_series_term_on_smp_architecture.md

Lines changed: 57 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,23 +2,23 @@
22

33
是时候开始行动了！ 让我们解决涉及在 SMP 架构中使用 PP 的多个输入的斐波那契数列的案例研究。 我正在使用配备双核处理器和四个线程的笔记本电脑。
44

5-
我们将为这个实现只导入两个模块，`os` 和 `pp.os` 模块将仅用于获取正在执行的进程的 `PID`。 我们将有一个名为 `input_list` 的列表，其中包含要计算的值和一个用于对结果进行分组的字典，我们将其称为 `result_dict`。 然后，我们转到代码块如下：
5+
我们将为这个实现只导入两个模块，`os` 和 `pp`。`os` 模块将仅用于获取正在执行的进程的 `PID`。 我们将有一个名为 `input_list` 的列表，其中包含要计算的值和一个用于对结果进行分组的字典，我们将其称为 `result_dict`。 然后，我们转到代码块如下：
66

77
```python
88
import os, pp
9+
910
input_list = [4, 3, 8, 6, 10]
1011
result_dict ={}
1112
```
1213

13-
然后，我们定义一个名为 `fibo_task` 的函数，它将由并行进程执行。 它将是我们通过 `Server` 类的提交方法传递的 `func` 参数。 该函数与前几章相比没有重大变化，除了现在通过使用元组封装参数中接收的值以及包含 `PID` 和计算的 `Fibonacci` 项的消息来完成返回。 看看下面的完整函数：
14+
然后，我们定义一个名为 `fibo_task` 的函数，它将由**并行进程**执行。 它将是我们通过 `Server` 类的提交方法传递的 `func` 参数。 该函数与前几章相比没有重大变化，除了现在通过使用元组封装参数中接收的值以及包含 `PID` 和计算的 `Fibonacci` 项的消息来完成返回。 看看下面的完整函数：
1415

1516
```python
1617
deffibo_task(value):
1718
a, b =0, 1
1819
for item inrange(value):
1920
a, b = b, a + b
20-
message ="the fibonacci calculated by pid %d was %d" \
21-
% (os.getpid(), a)
21+
message ="the fibonacci calculated by pid %d was %d"% (os.getpid(), a)
2222

2323
return (value, message)
2424
```
@@ -28,10 +28,11 @@ def fibo_task(value):
2828
```python
2929
defaggregate_results(result):
3030
print"Computing results with PID [%d]"% os.getpid()
31+
3132
result_dict[result[0]] = result[1]
3233
```
3334

34-
现在，我们有两个函数要定义。 我们必须创建一个实例用于分派任务的服务器类。 以下代码行创建一个服务器实例：
35+
现在，我们有两个函数要定义。 我们必须创建一个 `Server` 类的实例来分派任务。 以下代码行创建一个服务器实例：
3536

3637
```python
3738
job_server = pp.Server()
@@ -67,4 +68,54 @@ for key, value in result_dict.items():
6768

6869
以下屏幕截图说明了程序的输出：
6970

70-
![1](../imgs/6-01.png)
71+
```shell
72+
$ python feibonacci_pp_smp.py
73+
Computing results with PID [21058]
74+
Computing results with PID [21058]
75+
Computing results with PID [21058]
76+
Computing results with PID [21058]
77+
Computing results with PID [21058]
78+
Main process PID [21058]
79+
For input 4, the fibonacci calculated by pid 21059 was 3
80+
For input 3, the fibonacci calculated by pid 21060 was 2
81+
For input 6, the fibonacci calculated by pid 21062 was 8
82+
For input 8, the fibonacci calculated by pid 21061 was 21
83+
For input 10, the fibonacci calculated by pid 21065 was 55
84+
```
85+
86+
## 完整示例
87+
88+
译者注: `feibonacci_pp_smp.py`
89+
90+
```python
91+
import os, pp
92+
93+
input_list = [4, 3, 8, 6, 10]
94+
result_dict ={}
95+
96+
deffibo_task(value):
97+
a, b =0, 1
98+
for item inrange(value):
99+
a, b = b, a + b
100+
message ="the fibonacci calculated by pid %d was %d"% (os.getpid(), a)
101+
102+
return (value, message)
103+
104+
defaggregate_results(result):
105+
print("Computing results with PID [%d]"% os.getpid())
106+
107+
result_dict[result[0]] = result[1]
108+
109+
job_server = pp.Server()
110+
111+
for item in input_list:
112+
job_server.submit(fibo_task, (item,), modules=('os',), # 这里增加新的模块需要添加进来。
113+
callback=aggregate_results)
114+
115+
job_server.wait()
116+
117+
print("Main process PID [%d]"% os.getpid())
118+
119+
for key, value in result_dict.items():
120+
print("For input %d, %s"% (key, value))
121+
```

‎docs/chapter6/using_pp_to_make_a_distributed_web_crawler.md‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
# 使用PP创建分布式网络爬虫
2+
3+
现在我们已经使用 `PP` 并行执行代码来调度本地进程，现在是验证代码是否以分布式方式执行的时候了。 为此，我们将使用以下三台不同的机器：
4+
5+
- Iceman-Thinkpad-X220: Ubuntu 13.10
6+
- Iceman-Q47OC-500P4C： Ubuntu 12.04 LTS
7+
- Asgard-desktop: Elementary OS
8+
9+
我们将在如上列举的三台机器上测试pp组件在分布式环境下的使用。对此，我们实现了分布式网络爬虫。`web_crawler_pp_cluster.py`方法中，将`input_list`列举的URL分发到本地以及远端进程执行，`web_crawler_pp_cluster.py`中的回调函数将组织这些URL以及以及通过它们找到的前三个连接(URL)并进行分组。
10+
11+
让我们逐步分析代码以了解如何找到解决此问题的方法。 首先，我们将`import`必要的模块并定义要使用的数据结构。 与上一节一样，我们将创建一个 `input_list` 和一个包含最终处理结果的字典`result_dict`。 参考以下代码：
12+
13+
```python
14+
import os, re, requests, pp
15+
16+
ppurl_list = ['http://www.google.com/','http://gizmodo.uol.com.br/',
17+
'https://github.com/', 'http://br.search.yahoo.com/',
18+
'http://www.python.org/','http://www.python.org/psf/']
19+
20+
result_dict ={}
21+
```
22+
23+
现在，我们的 `aggregate_results` 函数将再次成为我们的回调函数，与斐波那契项的示例相比变化不大。 我们只是更改了要插入字典中的消息的格式，以及这个回调的返回将是一个包含执行它的进程的 `PID`、执行它的主机名和找到的前三个链接组成的元组。 参考`aggregate_results`函数如下：
24+
25+
```python
26+
defaggregate_results(result):
27+
print("Computing results in main process PID [%d]"% os.getpid())
28+
message ="PID %d in hostname [%s] the following links were found: %s"% (result[2], result[3], result[1])
29+
result_dict[result[0]] = message
30+
```
31+
32+
下一步是定义 `crawl_task` 函数，它将由 `Server` 类的实例调度。 该功能与前面章节中介绍的功能类似，旨在收集作为参数接收的 `URL` 显示的页面中的现有链接。 唯一的区别是返回是一个元组。 参考以下代码：
33+
34+
```python
35+
defcrawl_task(url):
36+
html_link_regex = re.compile('<a\s(?:.*?\s)*?href=[\'"](.*?)[\'"].*?>')
37+
request_data = requests.get(url)
38+
39+
links = html_link_regex.findall(request_data.text)[:3]
40+
return (url, links, os.getpid(), os.uname()[1])
41+
```
42+
43+
在`main`方法和`callback`方法定义之后，我们需要初始化`Server`类实例，以至于能够在分布式环境下执行网络爬虫任务。我们注意到`pp.Server`类有三个参数，第一个参数是执行`Server`类方法的`IP`或`hostname`，我们的例子中，除了本机之外，还需要定义另外两台机器的`IP`和`hostname`，定义如下所示：
44+
45+
```python
46+
ppservers = ("192.168.25.21", "192.168.25.9")
47+
```
48+
49+
!!! info ""
50+
51+
如果您不想通知并希望自动发现可用于接收任务的机器，请在 `ppservers` 元组中使用 `*` 字符串。
52+
53+
定义标识服务器的元组。 我们将创建一个 `Server` 实例，如下所示：
54+
55+
```python
56+
job_dispatcher = pp.Server(ncpus=1, ppservers=ppservers, socket_timeout=60000)
57+
```
58+
59+
值得注意的是，与前面的示例相比有一些变化。 首先，我们将值 `1` 传递给了 `ncpus` 参数。 这将导致 `PP` 创建单个本地进程，并在必要时将其他任务分派给远程机器。 定义的第二个参数是我们在上一步中创建的服务器的元组。 最后，我们为通信中使用的套接字定义了一个超时值，仅用于测试目的。 目标是避免网络超时而关闭通道。
60+
61+
创建 `Server` 类的实例后，就可以分派我们的函数来执行了。 让我们遍历每个 `URL` 并将它们传递给 `Server` 实例的`submit`方法，如下所示：
62+
63+
```python
64+
for url in ppurl_list:
65+
job_dispatcher.submit(crawl_task, (url,),
66+
modules=('os', 're', 'requests',),
67+
callback=aggregate_results)
68+
```
69+
70+
与之前计算斐波那契数列的示例相比，重要的变化是发送执行所需的模块。
71+
72+
!!! info ""
73+
74+
你一定在想为什么元组模块中没有传`PP`模块。 很简单; PP 执行环境已经为我们做了这个导入。 毕竟，它需要在远程节点中执行此操作。
75+
76+
为了完成我们的并行和分布式网络爬虫，我们必须等到执行结束才能显示它们的输出。 请注意，最后，`Server` 类的 `print_stats` 方法中有一个新元素，它显示了一些有趣的执行统计信息，如下所示：
77+
78+
```python
79+
job_dispatcher.wait()
80+
81+
print"\nMain process PID [%d]\n"% os.getpid()
82+
for key, value in result_dict.items():
83+
print"** For url %s, %s\n"% (key, value)
84+
job_dispatcher.print_stats()
85+
```
86+
87+
在执行程序之前，我们需要在远程机器上初始化 `ppserver.py` 实用程序； `ppserver.py –a –d` 是这里使用的命令，其中 `–a` 是自动发现的选项，允许未指定 `IP` 地址的客户端找到服务器。 另一个参数是 `-d`，它通过日志显示有关服务器活动如何执行的信息。
88+
89+
让我们按以下顺序可视化输出：
90+
91+
1. 首先，以下屏幕截图显示了主节点中创建和分发任务的阶段：
92+
![1](../imgs/6-02.png)
93+
2. 然后，初始化 `ppservers.py` 服务器并在以下屏幕截图中看到处理任务（从 iceman-Q47OC500P4C 的 `ppserver.py` 输出和从 `asgard-desktop` 的 `ppserver.py` 输出）。
94+
3. 在前面的屏幕截图中，值得注意的是统计信息带来了有趣的信息，例如分配到不同目的地的任务数量、每个任务的时间以及每个目的地的总数。 前面屏幕截图中的另一个相关点是回调函数仅在主进程中执行，即调度任务中的那些。 因此，切记不要让回调任务过重，因为它们可能会消耗主节点的过多资源； 这显然取决于每个案例的具体情况。
95+
4. 下面的截图是的`debug`模式下`ppserver.py`脚本在`iceman-Q470C-500P4C`机器上的的执行日志。
96+
97+
![1](../imgs/6-03.png)
98+
99+
5. 下面截图是`debug`模式下`ppserver.py`脚本在`asgard-desktop`机器上的执行日志。
100+
![1](../imgs/6-04.png)
101+
102+
## 完整示例
103+
104+
译者注: 由于机器性能原因或网络原因，未能还原作者执行场景。
105+
106+
`fibonacci_pp_cluster.py`
107+
108+
```python
109+
import os, re, requests, pp
110+
111+
112+
ppurl_list = ['http://www.google.com/','http://gizmodo.uol.com.br/',
113+
'https://github.com/', 'http://br.search.yahoo.com/',
114+
'http://www.python.org/','http://www.python.org/psf/']
115+
116+
result_dict ={}
117+
118+
defaggregate_results(result):
119+
print("Computing results in main process PID [%d]"% os.getpid())
120+
message ="PID %d in hostname [%s] the following links were found: %s"% (result[2], result[3], result[1])
121+
result_dict[result[0]] = message
122+
123+
defcrawl_task(url):
124+
html_link_regex = re.compile('<a\s(?:.*?\s)*?href=[\'"](.*?)[\'"].*?>')
125+
request_data = requests.get(url)
126+
127+
links = html_link_regex.findall(request_data.text)[:3]
128+
129+
return (url, links, os.getpid(), os.uname()[1])
130+
131+
132+
ppservers = ("192.168.25.21", "192.168.25.9")
133+
134+
job_dispatcher = pp.Server(ncpus=1, ppservers=ppservers, socket_timeout=60000)
135+
136+
for url in ppurl_list:
137+
job_dispatcher.submit(crawl_task, (url,),
138+
modules=('os', 're', 'requests', ),
139+
callback=aggregate_results)
140+
141+
job_dispatcher.wait()
142+
143+
print("\nMain process PID [%d]\n"% os.getpid())
144+
for key, value in result_dict.items():
145+
print("** For url %s, %s\n"% (key, value))
146+
job_dispatcher.print_stats() # 显示了一些有趣的执行统计信息
147+
```

‎docs/chapter6/使用pp模块创建分布式网络爬虫.md‎

Lines changed: 0 additions & 101 deletions
This file was deleted.

‎docs/chapter6/总结.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11

22
# 小结
33

4-
本章我们学习了使用更低层次的组件在没有直接关系的进程间通信。此外，我们介绍了PP模块，该模块提供本地进程和远程服务器进程通信的方法。PP模块用于创建简单，小型，并发和分布式的python应用。
4+
我们研究了使用更低级资源在进程之间建立没有直接关系的进程间通信。 此外，我们还了解了使用 `PP` 模块，这有助于我们抽象本地进程（包括分布式进程）之间的通信。 `PP` 是构建简单、小型、并行和分布式 Python 应用程序的便捷工具。
55

6-
下章我们将学习怎么使用Celery执行并发，分布式任务。
6+
在下一章中，我们将学习如何使用名为 `Celery` 的模块以并行和分布式的方式执行任务。

‎docs/imgs/6-01.png‎

-114 KB
Binary file not shown.

0 commit comments

Comments
(0)