python分布式爬虫及数据存储_Python简单分布式爬虫

你的名字 2022-10-25 14:26 227阅读 0赞

分布式爬虫采用主从模式。主从模式是指由一台主机作为控制节点，负责管理所有运行网络爬虫的主机(url管理器,数据存储器,控制调度器)，爬虫只需要从控制节点哪里接收任务，并把新生成任务提交给控制节点。此次使用三台主机进行分布式爬取,一台主机作为控制节点，另外两台主机作为爬虫节点。

控制节点主要分为url管理器、数据存储器和控制调度器。控制调度器通过三个进程来协调URL管理器和数据存储器的工作：一个是URL管理进程，负责URL的管理和将URL传递给爬虫节点,一个是数据提取进程,负责读取爬虫节点返回的数据,将返回数据中的URL交给URL管理进程，数据存储进程，负责将数据提取进程中提交的数据进行本地存储。

url管理器

\# coding:utf-8

try :

import cPickle as pickle

except ImportError:

import pickle

\#cPickle引用序列化包

import hashlib

class UrlManager(object):

def \_\_init\_\_(self):

self.new\_urls = self.load\_progress('new\_urls.txt') \# 未爬取URL集合

self.old\_urls = self.load\_progress('old\_urls.txt') \# 已爬取URL集合

def has\_new\_url(self):

\# 判断是否有未爬取的URL

return self.new\_url\_size() != 0

def get\_new\_url(self):

\# 获取一个未爬取的URL

new\_url = self.new\_urls.pop()

m = hashlib.md5()\#对url进行MD5加密

m.update(new\_url)

self.old\_urls.add(m.hexdigest()\[8:-8\])\#

return new\_url

def add\_new\_url(self, url):

\# 将新的URL添加到未爬取的URL结合中

if url is None:

return

m = hashlib.md5()

m.update(url)

url\_md5 = m.hexdigest()\[8:-8\]

if url not in self.new\_urls and url\_md5 not in self.old\_urls:

self.new\_urls.add(url) \# 将新的url添加到列表中

\# 批量添加url

def add\_new\_urls(self, urls):

\# 将新的URL添加到未爬取的URL集合中

if urls is None or len(urls) == 0:

return

for url in urls:

self.add\_new\_url(url)

\# 获取未爬取url集合的大小

def new\_url\_size(self):

return len(self.new\_urls)

\# 获取已经爬取URL集合的大小

def old\_url\_size(self):

return len(self.old\_urls)

\#保存进度

\#param path:文件路径

\#param data:数据

\# return:

def save\_progress(self, path, data):

with open(path, 'wb') as f:

pickle.dump(data, f)

\#从本地文件加载进度

\#param path 文件路径

\#return set集合

def load\_progress(self, path):

print '\[+\]从文件加载进度:%s' %path

try:

with open(path,'rb') as f:

tmp = pickle.load(f)

return tmp

except:

print '\[!\]无进度文件,创建:%s' % path

return set()

数据存储器

\# coding:utf-8

import codecs

import sys

import time

from urllib import unquote

class DataOutput(object):

def \_\_init\_\_(self):

self.filepath ='baike\_%s.html'%(time.strftime("%Y\_%m\_%d\_%H\_%M\_%S", time.localtime()))

self.output\_head(self.filepath)

self.datas = \[\]

def store\_data(self, data):

if data is None:

return

self.datas.append(data)

if len(self.datas)>10:

self.output\_html(self.filepath)

\#将HTML头写进去

\#param path:保存路径

def output\_head(self, path):

fout = codecs.open(path, 'w', encoding = 'uft-8')

fout.write("")

fout.write("

fout.close()

\#将数据写入HTML文件中

\#param path:文件路径

def output\_html(self,path):

fout = codecs.open(path, 'w', encoding = 'utf-8')

for data in self.datas:

fout.write("

%s" % data\['url'\])

fout.write("

%s" % data\['title'\])

fout.write("

%s" % data\['summary'\])

fout.write("

self.datas.remove(data)

fout.close()

\#输出HTML结束

\#param path文件存储路径

def output\_end(self,path):

fout = codecs.open(path, 'a', encoding = 'utf-8')

fout.write("

fout.write("")

fout.close()

控制调度器

\# coding:utf-8

import time, sys, Queue

import multiprocessing

from multiprocessing.managers import BaseManager

from UrlManager import UrlManager

from DataOutput import DataOutput

class QueueManager(BaseManager):

pass

class NodeManager(object):

\# 创建一个分布式管理器

\# param:url\_q url队列

\# param result\_q 结果队列

def start\_Manager(self, url\_q, result\_q):

\# 把创建的两个队列注册在网络上,利用register方法,callable参数关联了Queue对象

\# 将Queue对象在网络中暴露

QueueManager.register('get\_task\_queue', callable=lambda: url\_q)

QueueManager.register('get\_result\_queue', callable=lambda: result\_q)

\# 绑定端口8001,设置验证口令"baike"

manger = BaseManager(address=('', 8001), authkey='baike')

\# 返回manager对象

return manger

def url\_manager\_proc(self, url\_q, conn\_q, root\_url):

url\_manager = UrlManager()

url\_manager.add\_new\_url(root\_url)

while True:

\# 从URL管理器获取新的URL

while (url\_manager.has\_new\_url()):

new\_url = url\_manager.get\_new\_url()

\# 将新的URL发给工作节点

url\_q.put(new\_url)

print 'old\_url=', url\_manager.old\_url\_size()

if (url\_manager.old\_url\_size() > 2000):

\# 通知爬虫节点工作结束

url\_q.put('end')

print '控制节点发起结束通知!'

\# 关闭管理节点，同时存储set状态

url\_manager.save\_progress('new\_urls.txt', url\_manager.new\_urls)

url\_manager.save\_progress('old\_urls.txt', url\_manager.old\_urls)

return

try:

if not conn\_q.empty():

urls = conn\_q.get()

url\_manager.add\_new\_urls(urls)

except BaseException, e:

time.sleep(0.1) \# 延时休息

def result\_solve\_proc(self, result\_q, conn\_q, store\_q):

while (True):

try:

if not result\_q.empty():

content = result\_q.get(True)

if content\['new\_urls'\] == 'end':

print '结果分析进程接收通知然后结束!'

store\_q.put('end')

return

conn\_q.put(content\['new\_urls'\]) \# url为set类型

store\_q.put(content\['data'\]) \# 解析出来的数据为dict类型

else:

time.sleep(0.1) \# 延时休息

except BaseException, e:

time.sleep(0.1) \# 延时休息

def store\_proc(self, store\_q):

output = DataOutput()

while True:

if not store\_q.empty():

data = store\_q.get()

if data == 'end':

print '存储进程接受通知然后结束'

output.ouput\_end(output.filepath)

return

output.store\_data(data)

else:

time.sleep(0.1)

if \_\_name\_\_ == '\_\_main\_\_':

\# 初始化4个队列

url\_q = Queue.Queue()

result\_q = Queue.Queue()

store\_q = Queue.Queue()

conn\_q = Queue.Queue()

\# 创建分布式管理器

node = NodeManager()

manager = node.start\_Manager(url\_q, result\_q)

\# 创建URL管理进程、数据提取进程和数据存储进程

url\_manager\_proc = multiprocessing.Process(target=node.url\_manager\_proc, args=(url\_q, conn\_q, 'https://baike.baidu.com/item/%E7%BD%91%E7%BB%9C%E7%88%AC%E8%99%AB/5162711?fr=aladdin&fromid=22046949&fromtitle=%E7%88%AC%E8%99%AB'))

result\_solve\_proc = multiprocessing.Process(target=node.result\_solve\_proc, args=(result\_q, conn\_q, store\_q))

\# 启动3个进程和分布式管理器

url\_manager\_proc.start()

result\_solve\_proc.start()

manager.get\_server().serve\_forever()

HTML下载器

\# coding:utf-8

import requests

import urllib2

import sys

type = sys.getfilesystemencoding()

class HtmlDownloader(object):

def download(slef, url):

if url is None:

return None

user\_agent = 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36'

headers = \{'User-Agent': user\_agent\}

req = urllib2.Request(url, headers=headers)

response = urllib2.urlopen(req)

if response.getcode() == 200:

html = response.read().decode("UTF-8").encode(type)

return html

return None

HTML解析器

\# coding:utf-8

import re

import urlparse

from bs4 import BeautifulSoup

class HtmlParser(object):

\# page\_url下载页面的URL

\# html\_cont 下载的网页内容

\# 返回URL和数据

def parser(self, page\_url, html\_cont):

if page\_url is None or html\_cont is None:

return

soup = BeautifulSoup(html\_cont, 'html.parser')

new\_urls = self.\_get\_new\_urls(page\_url, soup)

new\_data = self.\_get\_new\_data(page\_url, soup)

return new\_urls, new\_data

\# page\_url下载页面的url

\# soup:soup

\# 返回新的URL集合

def \_get\_new\_urls(self, page\_url, soup):

new\_urls = set()

\# 抽取符合要求的a标记

links = soup.find\_all('a', href=re.compile(r'/item/.\*'))

for link in links:

\# 提取href属性

new\_url = link\['href'\]

\# 拼接成完整网址

new\_full\_url = urlparse.urljoin(page\_url, new\_url)

new\_urls.add(new\_full\_url)

return new\_urls

\# 下载页面的url

def \_get\_new\_data(self, page\_url, soup):

data = \{\}

data\['url'\] = page\_url

title = soup.find('dd', class\_='lemmaWgt-lemmaTitle-title').find('h1')

data\['title'\] = title.get\_text()

summary = soup.find('div', class\_='lemma-summary')

\# 获取tag中包含的所有文本内容,包括子孙tag中的内容,并将结果作为Unicode字符串返回

data\['summary'\] = summary.get\_text()

return data

爬虫调度器

\# coding:utf-8

import time, sys, Queue

from multiprocessing.managers import BaseManager

from UrlManager import UrlManager

from DataOutput import DataOutput

from HtmlDownloader import HtmlDownloader

from HtmlParser import HtmlParser

class SpoderWork(object):

def \_\_init\_\_(self):

\#初始化分布式进程中工作节点的连接工作

\#实现第一步：使用BaseManager注册用于获取Queue的方法名称

BaseManager.register('get\_task\_queue')

BaseManager.register('get\_result\_queue')

\#实现第二步:连接到服务器

server\_addr = '127.0.0.1'

print ('Connect to server %s....' % server\_addr)

self.m = BaseManager(address=(server\_addr,8001),authkey='baike')

\#从网络连接

self.m.connect()

\#实现第三步:获取Queue对象

self.task = self.m.get\_task\_queue()

self.result = self.m.get\_result\_queue()

\#初始化网页下载器和解析器

self.downloader = HtmlDownloader()

self.parser = HtmlParser()

print 'init finish'

def crawl(self):

while(True):

try:

if not self.task.empty():

url = self.task.get()

if url=='end':

print '控制节点通知爬虫节点停止工作'

self.result.put(\{'new\_urls':'end','data':'end'\})

return

print '爬虫节点正在解析:%s' % url.encode('utf-8')

content = self.downloader.download(url)

new\_urls,data = self.parser.parser(url,content)

self.result.put(\{'new\_urls': new\_urls, 'data': data\})

except EOFError,e:

print '连接工作节点失败'

return

except Exception,e:

print e

print 'Crawl fail'

if \_\_name\_\_=='\_\_main\_\_':

spider = SpoderWork()

spider.crawl()

python分布式爬虫及数据存储_Python简单分布式爬虫

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关爬虫--分布式爬虫

相关 Python之分布式爬虫

相关 Python爬虫学习记录——17.分布式爬虫

相关 Python爬虫之数据存储

相关 Python爬虫进阶——Scrapy框架原理及分布式爬虫构建

相关 python分布式爬虫及数据存储_Python简单分布式爬虫

相关 Python资料：python分布式爬虫打造搜索引擎

相关 Python 分布式爬虫

相关分布式爬虫

相关爬虫分布式