Skip to content

Commit 0105bc4

Browse files
author
Boris
committed
add batch spider
1 parent de8f3e5 commit 0105bc4

5 files changed

Lines changed: 356 additions & 0 deletions

File tree

‎docs/_sidebar.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
* [使用前必读](usage/使用前必读.md)
1212
* [AirSpider](usage/AirSpider.md)
1313
* [Spider](usage/Spider.md)
14+
* [BatchSpider](usage/BatchSpider.md)
1415

1516
* 框架剖析
1617
* [Request](source_code/Request.md)
@@ -20,6 +21,7 @@
2021
* [MysqlDB](source_code/MysqlDB.md)
2122
* [RedisDB](source_code/RedisDB.md)
2223
* [Spider进阶](source_code/Spider进阶.md)
24+
* [BatchSpider进阶](source_code/BatchSpider进阶.md)
2325
* [Item](source_code/Item.md)
2426
* [ItemBuffer](source_code/ItemBuffer.md)
2527

‎docs/source_code/BatchSpider进阶.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
# BatchSpider

‎docs/usage/BatchSpider.md‎

Lines changed: 310 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,310 @@
1+
# BatchSpider
2+
3+
BatchSpider是一款分布式批次爬虫框架，对于需要周期性采集的数据，优先考虑使用本爬虫。
4+
5+
## 1. 创建项目
6+
7+
参考 [Spider](usage/Spider?id=_1-创建项目)
8+
9+
10+
## 2. 创建爬虫
11+
12+
命令参考：[命令行工具](command/cmdline.md?id=_2-创建爬虫)
13+
14+
示例:
15+
16+
feapder create -s batch_spider_test 3
17+
18+
生成如下
19+
20+
```
21+
import feapder
22+
23+
24+
class BatchSpiderTest(feapder.BatchSpider):
25+
# 自定义数据库，若项目中有setting.py文件，此自定义可删除
26+
__custom_setting__ = dict(
27+
REDISDB_IP_PORTS="localhost:6379",
28+
REDISDB_USER_PASS="",
29+
REDISDB_DB=0,
30+
MYSQL_IP="localhost",
31+
MYSQL_PORT=3306,
32+
MYSQL_DB="feapder",
33+
MYSQL_USER_NAME="feapder",
34+
MYSQL_USER_PASS="feapder123",
35+
)
36+
37+
def start_requests(self, task):
38+
yield feapder.Request("https://www.baidu.com")
39+
40+
def parser(self, request, response):
41+
print(response)
42+
43+
44+
if __name__ == "__main__":
45+
spider = BatchSpiderTest(
46+
redis_key="xxx:xxxx", # redis中存放任务等信息的根key
47+
task_table="", # mysql中的任务表
48+
task_keys=["id", "xxx"], # 需要获取任务表里的字段名，可添加多个
49+
task_state="state", # mysql中任务状态字段
50+
batch_record_table="xxx_batch_record", # mysql中的批次记录表
51+
batch_name="xxx(周全)", # 批次名字
52+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
53+
)
54+
55+
# spider.start_monitor_task() # 下发及监控任务
56+
spider.start() # 采集
57+
```
58+
59+
因BatchSpider是基于redis做的分布式，mysql来维护任务种子及批次信息，因此模板代码默认给了redis及mysql的配置方式，连接信息需按真实情况修改
60+
61+
## 3. 代码讲解
62+
63+
配置信息：
64+
65+
- REDISDB_IP_PORTS： 连接地址，若为集群或哨兵模式，多个连接地址用逗号分开，若为哨兵模式，需要加个REDISDB_SERVICE_NAME参数
66+
- REDISDB_USER_PASS： 连接密码
67+
- REDISDB_DB：数据库
68+
69+
BatchSpider参数：
70+
71+
1. redis_key：redis中存储任务等信息的key前缀，如redis_key="feapder:spider_test", 则redis中会生成如下
72+
73+
![-w365](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/02/21/16139009217536.jpg?x-oss-process=style/markdown-media)
74+
75+
1. task_table：mysql中的任务表，为抓取的任务种子
76+
2. task_keys：任务表里需要获取的字段，框架会将这些字段的数据查询出来，传递给爬虫，然后拼接请求
77+
3. task_state：任务表里表示任务完成状态的字段，默认是state。字段为整形，有4种状态（0 待抓取，1抓取完毕，2抓取中，-1抓取失败）
78+
4. batch_record_table：批次信息表，用于记录批次信息
79+
5. batch_name： 批次名称，可以理解成爬虫的名字，用于报警等
80+
6. batch_interval：批次周期 天为单位 若为小时 可写 1 / 24
81+
82+
启动：BatchSpider分为master及work两种程序
83+
84+
1. master负责下发任务，监控批次进度，创建批次等功能，启动方式：
85+
86+
spider.start_monitor_task()
87+
88+
2. worker负责消费任务，抓取数据，启动方式：
89+
90+
spider.start()
91+
92+
93+
更详细的说明可查看 [BatchSpider进阶](source_code/BatchSpider进阶.md)
94+
95+
## 4. 声明
96+
97+
[Spider](usage/Spider.md)支持的方法BatchSpider都支持，使用方式一致，下面重点讲解不同之处
98+
99+
## 5. 任务表
100+
101+
任务表为存储任务种子的，表结构需要包含`id`、`任务状态`两个字段，如我们需要对某些地址进行采集，设计如下
102+
103+
![-w752](media/16139762922842.jpg)
104+
105+
建表语句：
106+
107+
```
108+
CREATE TABLE `batch_spider_task` (
109+
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
110+
`url` varchar(255) DEFAULT NULL,
111+
`state` int(11) DEFAULT '0',
112+
PRIMARY KEY (`id`)
113+
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
114+
```
115+
116+
也许有人会问，为什么要弄个任务表，直接把种子任务写到代码里不行么。答：可以的，可以用`AirSpider`或`Spider`这么搞。`BatchSpider`面向的场景是周期性抓取，如我们有1亿个商品需要更新，不可能把这1亿个商品id都写代码里，还是需要存储到一张表里，这个表即为任务表。
117+
118+
为了保证每个商品都得以更新，需要引入抓取状态字段，本例为`state`字段。`state`字段有4种状态（0 待抓取，1抓取完毕，2抓取中，-1抓取失败）。框架下发任务时，会优先分批下发状态为0的任务到redis任务队列，并将这些已下发的任务状态更新为2，当0都下发完毕且redis任务队列中无任务，这时框架会检查任务表里是否还有状态为2的任务，若有则将这些任务视为丢失的任务，然后将这些状态为2的任务置为0，再次分批下发到redis任务队列。直到任务表里任务状态只有1和-1两种状态，才算采集完毕
119+
120+
1 和 -1 两种状态是开发人员在代码里自己维护的。当任务做完时将任务状态更新为1，当任务无效时，将任务状态更新为-1。更新方法见[更新任务状态](usage/BatchSpider?id=_8-更新任务状态)
121+
122+
注意：每个批次开始时，框架默认会重置状态非-1的任务为0，然后重新抓取。-1的任务永远不会抓取
123+
124+
## 7. 拼接任务
125+
126+
def start_requests(self, task):
127+
pass
128+
129+
任务拼接在`start_requests`里处理。这里的task参数为元组，值为BatchSpider启动参数中指定的`task_keys`对应的值
130+
131+
如表`batch_spider_task`，现有任务信息如下：
132+
![-w398](media/16139773315622.jpg)
133+
134+
启动参数配置如下，注意`task_keys=["id", "url"]`：
135+
136+
```
137+
def crawl_test(args):
138+
spider = test_spider.TestSpider(
139+
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
140+
task_table="batch_spider_task", # mysql中的任务表
141+
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
142+
task_state="state", # mysql中任务状态字段
143+
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
144+
batch_name="批次爬虫测试(周全)", # 批次名字
145+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
146+
)
147+
148+
if args == 1:
149+
spider.start_monitor_task() # 下发及监控任务
150+
else:
151+
spider.start() # 采集
152+
```
153+
154+
这时，start_requests的task参数值即为任务表里id与url对应的值。
155+
156+
```
157+
158+
def start_requests(self, task):
159+
# task 为在任务表中取出的每一条任务
160+
id, url = task # id， url为所取的字段，main函数中指定的
161+
yield feapder.Request(url, task_id=id)
162+
163+
```
164+
165+
## 8. 更新任务状态
166+
167+
任务的完成状态与失败状态需要自己维护，为了更新这个状态，我们需要在请求中携带任务id，常规写法为
168+
169+
yield feapder.Request(url, task_id=id)
170+
171+
当任务解析完毕后，可使用如下方法更新
172+
173+
yield self.update_task_batch(request.task_id, 1) # 更新任务状态为1
174+
175+
这个更新不是实时的，也会先流经ItemBuffer，然后在数据入库后批量更新
176+
177+
## 9. 处理无效任务
178+
179+
有些任务，可能就是有问题的，我们需要将其更新为-1，防止爬虫一直重试。除了在解析函数中判断当前任务是否有效外，框架还提供了两个函数
180+
181+
```
182+
def exception_request(self, request, response):
183+
"""
184+
@summary: 请求或者parser里解析出异常的request
185+
---------
186+
@param request:
187+
@param response:
188+
---------
189+
@result: request / callback / None (返回值必须可迭代)
190+
"""
191+
192+
pass
193+
194+
def failed_request(self, request, response):
195+
"""
196+
@summary: 超过最大重试次数的request
197+
---------
198+
@param request:
199+
---------
200+
@result: request / item / callback / None (返回值必须可迭代)
201+
"""
202+
203+
pass
204+
```
205+
206+
`exception_request`：处理请求失败或解析出异常的request，我们可以在这里切换request的cookie等，然后再`yield request`返回处理后的request
207+
208+
`failed_request`：处理超过最大重试次数的request。我们可以在这里将任务状态更新为-1
209+
210+
def failed_request(self, request, response):
211+
"""
212+
@summary: 超过最大重试次数的request
213+
---------
214+
@param request:
215+
---------
216+
@result: request / item / callback / None (返回值必须可迭代)
217+
"""
218+
219+
yield request
220+
yield self.update_task_batch(request.task_id, -1) # 更新任务状态为-1
221+
222+
超过最大重试次数的request会保存到redis里，key名以`z_failed_requsets`结尾。我们可以查看这个表里的失败任务，观察失败原因，以此来调整爬虫
223+
224+
## 10.增量采集
225+
226+
每个批次开始时，框架默认会重置状态非-1的任务为0，然后重新抓取。但是有些需求是增强采集的，做过的任务无需再次处理。重置任务是`init_task`方法实现的，我们可以将`init_task`方法置空来实现增量采集
227+
228+
```
229+
def init_task(self):
230+
pass
231+
```
232+
233+
## 11. 调试
234+
235+
与[Spider调试](usage/Spider?id=_6-调试)类似。BatchSpider可以通过`to_DebugBatchSpider`转为调试爬虫，写法如下：
236+
237+
```
238+
def test_debug():
239+
spider = test_spider.TestSpider.to_DebugBatchSpider(
240+
task_id=1,
241+
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
242+
task_table="batch_spider_task", # mysql中的任务表
243+
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
244+
task_state="state", # mysql中任务状态字段
245+
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
246+
batch_name="批次爬虫测试(周全)", # 批次名字
247+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
248+
)
249+
250+
spider.start() # 采集
251+
```
252+
253+
DebugBatchSpider爬虫支持传递`task_id`或直接传递`task`来指定任务。还支持其他参数，全部参数如下：
254+
255+
@param task_id: 任务id
256+
@param task: 任务 task 与 task_id 二者选一即可
257+
@param save_to_db: 数据是否入库 默认否
258+
@param update_stask: 是否更新任务 默认否
259+
260+
261+
## 12. 运行BatchSpider
262+
263+
与[Spider](usage/Spider?id=_7-运行多个spider)运行方式类型。但因每个爬虫都有maser和work两个入口，因此框架提供一种更方便的方式，写法如下
264+
265+
```
266+
from spiders import *
267+
from feapder import ArgumentParser
268+
269+
270+
def crawl_test(args):
271+
spider = test_spider.TestSpider(
272+
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
273+
task_table="batch_spider_task", # mysql中的任务表
274+
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
275+
task_state="state", # mysql中任务状态字段
276+
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
277+
batch_name="批次爬虫测试(周全)", # 批次名字
278+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
279+
)
280+
281+
if args == 1:
282+
spider.start_monitor_task() # 下发及监控任务
283+
else:
284+
spider.start() # 采集
285+
286+
if __name__ == "__main__":
287+
288+
parser = ArgumentParser(description="批次爬虫测试")
289+
290+
parser.add_argument(
291+
"--crawl_test", type=int, nargs=1, help="BatchSpider demo(1|2）", function=crawl_test
292+
)
293+
294+
parser.start()
295+
296+
```
297+
298+
运行master
299+
300+
python3 main.py --crawl_test 1
301+
302+
运行worker
303+
304+
python3 main.py --crawl_test 2
305+
306+
crawl_test的args参数会接收1或2两个参数，以此来运行不同的程序
307+
308+
## 13. 完整的代码示例
309+
310+
[https://github.com/Boris-code/feapder/tree/master/tests/batch-spider](https://github.com/Boris-code/feapder/tree/master/tests/batch-spider)

‎tests/batch-spider/main.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,21 @@ def crawl_test(args):
2727
else:
2828
spider.start() # 采集
2929

30+
def test_debug():
31+
spider = test_spider.TestSpider.to_DebugBatchSpider(
32+
task_id=1,
33+
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
34+
task_table="batch_spider_task", # mysql中的任务表
35+
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
36+
task_state="state", # mysql中任务状态字段
37+
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
38+
batch_name="批次爬虫测试(周全)", # 批次名字
39+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
40+
)
41+
42+
43+
spider.start() # 采集
44+
3045

3146
if __name__ == "__main__":
3247

@@ -35,6 +50,7 @@ def crawl_test(args):
3550
parser.add_argument(
3651
"--crawl_test", type=int, nargs=1, help="(1|2）", function=crawl_test
3752
)
53+
parser.add_argument("--test_debug", action="store_true", help="测试debug", function=test_debug)
3854

3955
parser.start()
4056

‎tests/batch-spider/spiders/test_spider.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,9 @@
1212

1313

1414
class TestSpider(feapder.BatchSpider):
15+
# def init_task(self):
16+
# pass
17+
1518
def start_requests(self, task):
1619
# task 为在任务表中取出的每一条任务
1720
id, url = task # id， url为所取的字段，main函数中指定的
@@ -24,4 +27,28 @@ def parser(self, request, response):
2427
yield item # 返回item， item会自动批量入库
2528
yield self.update_task_batch(request.task_id, 1) # 更新任务状态为1
2629

30+
def exception_request(self, request, response):
31+
"""
32+
@summary: 请求或者parser里解析出异常的request
33+
---------
34+
@param request:
35+
@param response:
36+
---------
37+
@result: request / callback / None (返回值必须可迭代)
38+
"""
39+
40+
pass
41+
42+
def failed_request(self, request, response):
43+
"""
44+
@summary: 超过最大重试次数的request
45+
---------
46+
@param request:
47+
---------
48+
@result: request / item / callback / None (返回值必须可迭代)
49+
"""
50+
51+
yield request
52+
yield self.update_task_batch(request.task_id, -1) # 更新任务状态为-1
53+
2754

0 commit comments

Comments
 (0)