开发者社区> seancheney> 正文

《Learning Scrapy》(中文版)第9章 使用Pipelines

简介: 序言第1章 Scrapy介绍第2章 理解HTML和XPath第3章 爬虫基础 第4章 从Scrapy到移动应用第5章 快速构建爬虫第6章 Scrapinghub部署第7章 配置和管理第8章 Scrapy编程 第9章 使用Pipeline第10章 理解Scrapy的性能第11章(完) Scrapyd分布式抓取和实时分析 在上一章,我们学习了如何辨析Scrapy中间件。
+关注继续查看

序言
第1章 Scrapy介绍
第2章 理解HTML和XPath
第3章 爬虫基础
第4章 从Scrapy到移动应用
第5章 快速构建爬虫
第6章 Scrapinghub部署
第7章 配置和管理
第8章 Scrapy编程
第9章 使用Pipeline
第10章 理解Scrapy的性能
第11章(完) Scrapyd分布式抓取和实时分析


在上一章,我们学习了如何辨析Scrapy中间件。在本章中,我们通过实例学习编写pipelines,包括使用REST APIs、连接数据库、处理CPU密集型任务、与老技术结合。

我们在本章中会使用集中新的数据库,列在下图的右边:

img_4366c15ba3755899dcce1803020b50a1.png

Vagrant已经配置好了数据库,我们可以从开发机向其发送ping,例如ping es或ping mysql。让我们先来学习REST APIs。

使用REST APIs

REST是用来一套创建网络服务的技术集合。它的主要优点是,比起SOAP和专有web服务,REST更简单和轻量。软件开发者注意到了web服务的CRUD(Create、Read、Update、Delete)和HTTP操作(GET、POST、PUT、DELETE)的相似性。它们还注意到传统web服务调用需要的信息可以再URL源进行压缩。例如,http://api.mysite.com/customer/john是一个URL源,它可以让我们分辨目标服务器,,更具体的,名字是john的服务器(行的主键)。它与其它技术结合时,比如安全认证、无状态服务、缓存、输出XML或JSON时,可以提供一个强大但简单的跨平台服务。REST席卷软件行业并不奇怪。

Scrapy pipeline的功能可以用REST API来做。接下来,我们来学习它。

使用treq

treq是一个Python包,它在Twisted应用中和Python的requests包相似。它可以让我们做出GET、POST、和其它HTTP请求。可以使用pip install treq安装,开发机中已经安装好了。

比起Scrapy的Request/crawler.engine.download() API,我们使用treq,因为后者具有性能优势,详见第10章。

一个写入Elasticsearch的pipeline

我们从一个向ES服务器(Elasticsearch)写入Items的爬虫开始。你可能觉得从ES开始,而不是MySQL,有点奇怪,但实际上ES是最容易的。ES可以是无模式的,意味着我们可以不用配置就使用它。treq也足以应付需要。如果想使用更高级的ES功能,我们应该使用txes2和其它Python/Twisted ES包。

有了Vagrant,我们已经有个一个运行的ES服务器。登录开发机,验证ES是否运行:

$ curl http://es:9200
{
  "name" : "Living Brain",
  "cluster_name" : "elasticsearch",
  "version" : { ... },
  "tagline" : "You Know, for Search"
}

在浏览器中登录http://localhost:9200也可以看到相同的结果。如果访问http://localhost:9200/properties/property/_search,我们可以看到一个响应,说ES已经进行了全局尝试,但是没有找到索引页。

笔记:在本章中,我们会在项集合中插入新的项,如果你想恢复原始状态的话,可以用下面的命令:

$ curl -XDELETE http://es:9200/properties

本章中的pipeline完整代码还有错误处理的功能,但我尽量让这里的代码简短,以突出重点。

提示:本章位于目录ch09,这个例子位于ch09/properties/properties/pipelines/es.py。

本质上,这个爬虫只有四行:

@defer.inlineCallbacks
def process_item(self, item, spider):
    data = json.dumps(dict(item), ensure_ascii=False).encode("utf- 8")
    yield treq.post(self.es_url, data)

前两行定义了一个标准process_item()方法,它可以产生延迟项。(参考第8章)

第三行准备了插入的data。ensure_ascii=False可使结果压缩,并且没有跳过非ASCII字符。我们然后将JSON字符串转化为JSON标准的默认编码UTF-8。

最后一行使用了treq的post()方法,模拟一个POST请求,将我们的文档插入ElasticSearch。es_url,例如http://es:9200/properties/property存在settings.py文件中(ES_PIPELINE_URL设置),它提供重要的信息,例如我们想要写入的ES的IP和端口(es:9200)、集合名(properties)和对象类型(property)。

为了是pipeline生效,我们要在settings.py中设置ITEM_PIPELINES,并启动ES_PIPELINE_URL设置:

ITEM_PIPELINES = {
    'properties.pipelines.tidyup.TidyUp': 100,
    'properties.pipelines.es.EsWriter': 800,
}
ES_PIPELINE_URL = 'http://es:9200/properties/property'

这么做完之后,我们前往相应的目录:

$ pwd
/root/book/ch09/properties
$ ls
properties  scrapy.cfg

然后运行爬虫:

$ scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=90
...
INFO: Enabled item pipelines: EsWriter...
INFO: Closing spider (closespider_itemcount)...
   'item_scraped_count': 106,

如果现在访问http://localhost:9200/properties/property/_search,除了前10条结果,我们可以在响应的hits/total字段看到插入的文件数。我们还可以添加参数?size=100以看到更多的结果。通过添加q= URL搜索中的参数,我们可以在全域或特定字段搜索关键词。相关性最强的结果会首先显示出来。例如,http://localhost:9200/properties/property/_search?q=title:london,可以让标题变为London。对于更复杂的查询,可以在https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html查询ES文档。

ES不需要配置,因为它根据提供的第一个文件,进行模式(字段类型)自动检测的。通过访问http://localhost:9200/properties/,我们可以看到它自动检测的映射。

再次运行crawl easy -s CLOSESPIDER_ITEMCOUNT=1000。因为pipelines的平均时间从0.12变为0.15秒,平均延迟从0.78变为0.81秒。吞吐量仍保持每秒约25项。

笔记:用pipelines向数据库插入Items是个好方法吗?答案是否定的。通常来讲,数据库更简单的方法以大量插入数据,我们应该使用这些方法大量批次插入数据,或抓取完毕之后进行后处理。我们会在最后一章看到这些方法。然后,还是有很多人使用pipelines向数据库插入文件,相应的就要使用Twisted APIs。

pipeline使用Google Geocoding API进行地理编码

我们的房子有各自所在的区域,我们还想对它们进行地理编码,即找到相应的坐标(经度、纬度)。我们可以将坐标显示在地图上,或计算距离。建这样的数据库需要复杂的数据库、复杂的文本匹配,还有复杂的空间计算。使用Google Geocoding API,我们可以避免这些。在浏览器中打开它,或使用curl取回以下URL的数据:

$ curl "https://maps.googleapis.com/maps/api/geocode/json?sensor=false&ad
dress=london"
{
   "results" : [
         ...
         "formatted_address" : "London, UK",
         "geometry" : {
            ...
            "location" : {
               "lat" : 51.5073509,
               "lng" : -0.1277583
          },
            "location_type" : "APPROXIMATE",
            ...
   ],
   "status" : "OK"
}

我们看到一个JSON对象,如果搜索一个location,我们可以快速获取伦敦中心的坐标。如果继续搜索,我们可以看到相同文件中海油其它地点。第一个是相关度最高的。因此如果存在results[0].geometry.location的话,它就是我们要的结果。

可以用前面的方法(treq)使用Google Geocoding API。只需要几行,我们就可以找到一个地址的坐标(目录pipelines中的geo.py),如下所示:

@defer.inlineCallbacks
def geocode(self, address):
   endpoint = 'http://web:9312/maps/api/geocode/json'
   parms = [('address', address), ('sensor', 'false')]
   response = yield treq.get(endpoint, params=parms)
   content = yield response.json()
   geo = content['results'][0]["geometry"]["location"]
   defer.returnValue({"lat": geo["lat"], "lon": geo["lng"]})

这个函数做出了一条URL,但我们让它指向一个可以离线快速运行的假程序。你可以使用endpoint = 'https://maps.googleapis.com/maps/api/geocode/json'连接Google服务器,但要记住它对请求的限制很严格。address和sensor的值是URL自动编码的,使用treq的方法get()的参数params。对于第二个yield,即response.json(),我们必须等待响应主题完全加载完毕对解析为Python对象。此时,我们就可以找到第一个结果的地理信息,格式设为dict,使用defer.returnValue()返回,它使用了inlineCallbacks。如果发生错误,这个方法会扔出例外,Scrapy会向我们报告。

通过使用geocode(),process_item()变成了一行语句:

item["location"] = yield self.geocode(item["address"][0])

设置让pipeline生效,将它添加到ITEM_PIPELINES,并设定优先数值,该数值要小于ES的,以让ES获取坐标值:

ITEM_PIPELINES = {
    ...
    'properties.pipelines.geo.GeoPipeline': 400,

开启数据调试,然后运行:

$ scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=90 -L DEBUG
...
{'address': [u'Greenwich, London'],
...
 'image_URL': [u'http://web:9312/images/i06.jpg'],
 'location': {'lat': 51.482577, 'lon': -0.007659},
 'price': [1030.0],
...

我们现在可以看到Items里的location字段。如果使用真正的Google API的URL运行,会得到例外:

File "pipelines/geo.py" in geocode (content['status'], address))
Exception: Unexpected status="OVER_QUERY_LIMIT" for  
address="*London"

这是为了检查我们在完整代码中插入了地点,以确保Geocoding API响应的status字段有OK值。除非是OK,否则我们取回的数据不会有设定好的格式,进而不能使用。对于这种情况,我们会得到OVER_QUERY_LIMIT状态,它指明我们在某处做错了。这个问题很重要,也很常见。应用Scrapy的高性能引擎,进行缓存、限制请求就很必要了。

我们可以在Geocoder API的文档,查看它的限制,“每24小时,免费用户可以进行2500次请求,每秒5次请求”。即使我们使用付费版本,仍有每秒10次请求的限制,所以这里的分析是有意义的。

笔记:后面的代码看起来可能有些复杂,复杂度还要取决于实际情况。在多线程环境中创建这样的组件,需要线程池和同步,这样代码就会变复杂。

这是一个简易的运用Twisted技术的限制引擎:

class Throttler(object):
    def __init__(self, rate):
        self.queue = []
        self.looping_call = task.LoopingCall(self._allow_one)
        self.looping_call.start(1. / float(rate))
    def stop(self):
        self.looping_call.stop()
    def throttle(self):
        d = defer.Deferred()
        self.queue.append(d)
        return d
    def _allow_one(self):
        if self.queue:
            self.queue.pop(0).callback(None)

这可以让延迟项在一个列表中排队,逐个触发,调用_allow_one();_allow_one()检查队列是否为空,如果不是,它会调用第一个延迟项的callback()。我们使用Twisted的task.LoopingCall() API,周期性调用_allow_one()。使用Throttler很容易。我们在pipeline的init初始化它,当爬虫停止时清空它:

class GeoPipeline(object):
    def __init__(self, stats):
        self.throttler = Throttler(5)  # 5 Requests per second
    def close_spider(self, spider):
        self.throttler.stop()

在使用限定源之前,我们的例子是在process_item()中调用geocode(),必须yield限制器的throttle()方法:

yield self.throttler.throttle()
item["location"] = yield self.geocode(item["address"][0])

对于第一个yield,代码会暂停一下,一段时间之后,会继续运行。例如,当某时有11个延迟项时,限制是每秒5次请求,即时间为11/5=2.2秒之后,队列变空,代码会继续。

使用Throttler,不再有错误,但是爬虫会变慢。我们看到示例中的房子只有几个不同的地址。这时使用缓存非常好。我们使用一个简单的Python dict来做,但这么可能会有竞争条件,这样会造成伪造的API请求。下面是一个没有此类问题的缓存方法,展示了Python和Twisted的特点:

class DeferredCache(object):
    def __init__(self, key_not_found_callback):
        self.records = {}
        self.deferreds_waiting = {}
        self.key_not_found_callback = key_not_found_callback
    @defer.inlineCallbacks
    def find(self, key):
        rv = defer.Deferred()
        if key in self.deferreds_waiting:
            self.deferreds_waiting[key].append(rv)
        else:
            self.deferreds_waiting[key] = [rv]
            if not key in self.records:
                try:
                    value = yield self.key_not_found_callback(key)
                    self.records[key] = lambda d: d.callback(value)
                except Exception as e:
                    self.records[key] = lambda d: d.errback(e)
            action = self.records[key]
            for d in self.deferreds_waiting.pop(key):
                reactor.callFromThread(action, d)
        value = yield rv
        defer.returnValue(value)

这个缓存看起来有些不同,它包含两个组件:

  • self.deferreds_waiting:这是一个延迟项的队列,等待给键赋值
  • self.records:这是键值对中出现过的dict

在find()方法的中间,如果没有在self.records找到一个键,我们会调用预先定义的callback函数,以取回丢失的值(yield self.key_not_found_callback(key))。这个调回函数可能会扔出一个例外。如何在Python中压缩存储值或例外呢?因为Python是一种函数语言,根据是否有例外,我们在self.records中保存小函数(lambdas),调用callback或errback。lambda函数定义时,就将值或例外附着在上面。将变量附着在函数上称为闭包,闭包是函数语言最重要的特性之一。

笔记:缓存例外有点不常见,但它意味着首次查找key时,key_not_found_callback(key)返回了一个例外。当后续查找还找这个key时,就免去了调用,再次返回这个例外。

find()方法其余的部分提供了一个避免竞争条件的机制。如果查找某个键已经在进程中,会在self.deferreds_waiting dict中有记录。这时,我们不在向key_not_found_callback()发起另一个调用,只是在延迟项的等待列表添加这个项。当key_not_found_callback()返回时,键有了值,我们触发所有的等待这个键的延迟项。我们可以直接发起action(d),而不用reactor.callFromThread(),但需要处理每个扔给下游的例外,我们必须创建不必要的很长的延迟项链。

使用这个缓存很容易。我们在init()对其初始化,设定调回函数为API调用。在process_item()中,使用缓存查找的方法如下:

def __init__(self, stats):
    self.cache = DeferredCache(self.cache_key_not_found_callback)
@defer.inlineCallbacks
def cache_key_not_found_callback(self, address):
    yield self.throttler.enqueue()
    value = yield self.geocode(address)
    defer.returnValue(value)
@defer.inlineCallbacks
def process_item(self, item, spider):
    item["location"] = yield self.cache.find(item["address"][0])
    defer.returnValue(item)

提示:完整代码位于ch09/properties/properties/pipelines/geo2.py。

为了使pipeline生效,我们使前一个方法无效,并添加当前的到settings.py的ITEM_PIPELINES:

ITEM_PIPELINES = {
    'properties.pipelines.tidyup.TidyUp': 100,
    'properties.pipelines.es.EsWriter': 800,
    # DISABLE 'properties.pipelines.geo.GeoPipeline': 400,
    'properties.pipelines.geo2.GeoPipeline': 400,
}

运行爬虫,用如下代码:

$ scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=1000
...
Scraped... 15.8 items/s, avg latency: 1.74 s and avg time in pipelines: 
0.94 s
Scraped... 32.2 items/s, avg latency: 1.76 s and avg time in pipelines: 
0.97 s
Scraped... 25.6 items/s, avg latency: 0.76 s and avg time in pipelines: 
0.14 s
...
: Dumping Scrapy stats:...
   'geo_pipeline/misses': 35,
   'item_scraped_count': 1019,

当填充缓存时,我们看到抓取的延迟变高。缓存结束时,延迟降低。数据还显示有35个遗漏,正好是数据集中不同地点的数目。很明显,上例中一共有1019 - 35= 984次API请求。如果我们使用真正的Google API,并提高每秒的API请求数,例如通过改变Throttler(5)到Throttler(10),使从5提高到10,我们可以将重试添加到geo_pipeline/retries stat记录中。如果有错误的话,例如,使用API找不到某个地点,会扔出一个例外,这会被geo_pipeline/errors stat记录。如果地点通过什么方式已经存在了,会在geo_pipeline/already_set stat中指明。最后,如果我们访问http://localhost:9200/properties/property/_search,以检查ES中的房子,我们可以看到包括地点的记录,例如{..."location": {"lat": 51.5269736, "lon": -0.0667204}...}。(运行前确保清空集合,去除旧的值)

在Elasticsearch进行地理索引

我们已经有了地点,我们可以将它们按距离排序。下面是一个HTTP POST请求,返回标题中包含Angel的房子,按照离点{51.54, -0.19}的距离进行排序:

$ curl http://es:9200/properties/property/_search -d '{
    "query" : {"term" : { "title" : "angel" } },
    "sort": [{"_geo_distance": {
        "location":      {"lat":  51.54, "lon": -0.19},
        "order":         "asc",
        "unit":          "km", 
        "distance_type": "plane" 
}}]}'

唯一的问题是如果我们运行它,我们会看到一个错误信息"failed to find mapper for [location] for geo distance based sort"。它指出,我们的location字段没有正确的空间计算的格式。为了设定正确的格式,我们要手动覆盖默认格式。首先,我们将自动检测的映射保存起来,将它作为起点:

$ curl 'http://es:9200/properties/_mapping/property' > property.txt

然后,我们如下所示编辑property.txt:

"location":{"properties":{"lat":{"type":"double"},"lon":{"type":"double"}}}

我们将这行代码替换为:

"location": {"type": "geo_point"}

我们还在文件最后删除了{"properties":{"mappings": and two }}。文件现在就处理完了。我们现在可以删除旧的类型,并用下面的schema建立新的类型:

$ curl -XDELETE 'http://es:9200/properties'
$ curl -XPUT 'http://es:9200/properties'
$ curl -XPUT 'http://es:9200/properties/_mapping/property' --data  
@property.txt

我们现在可以用之前的命令,进行一个快速抓取,将结果按距离排序。我们的搜索返回的是房子的JSONs对象,其中包括一个额外的sort字段,显示房子离某个点的距离。

连接数据库与Python客户端

可以连接Python Database API 2.0的数据库有许多种,包括MySQL、PostgreSQL、Oracle、Microsoft、SQL Server和SQLite。它们的驱动通常很复杂且进行过测试,为Twisted再进行适配会浪费很多时间。可以在Twisted应用中使用数据库客户端,例如,Scrapy可以使用twisted.enterprise.adbapi库。我们使用MySQL作为例子,说明用法,原则也适用于其他数据库。

用pipeline写入MySQL

MySQL是一个好用又流行的数据库。我们来写一个pipeline,来向其中写入文件。我们的虚拟环境中,已经有了一个MySQL实例。我们用MySQL命令行来做一些基本的管理操作,命令行工具已经在开发机中预先安装了:

$ mysql -h mysql -uroot -ppass

mysql>提示MySQL已经运行,我们可以建立一个简单的含有几个字段的数据表,如下所示:

mysql> create database properties;
mysql> use properties
mysql> CREATE TABLE properties (
  url varchar(100) NOT NULL,
  title varchar(30),
  price DOUBLE,
  description varchar(30),
  PRIMARY KEY (url)
);
mysql> SELECT * FROM properties LIMIT 10;
Empty set (0.00 sec)

很好,现在已经建好了一个包含几个字段的MySQL数据表,它的名字是properties,可以开始写pipeline了。保持MySQL控制台打开,我们过一会儿会返回查看是否有差入值。输入exit,就可以退出。

笔记:在这一部分中,我们会向MySQL数据库插入properties。如果你想删除,使用以下命令:

mysql> DELETE FROM properties;

我们使用MySQL的Python客户端。我们还要安装一个叫做dj-database-url的小功能模块(它可以帮我们设置不同的IP、端口、密码等等)。我们可以用pip install dj-database-url MySQL-python,安装这两项。我们的开发机上已经安装好了。我们的MySQL pipeline很简单,如下所示:

from twisted.enterprise import adbapi
...
class MysqlWriter(object):
    ...
    def __init__(self, mysql_url):
        conn_kwargs = MysqlWriter.parse_mysql_url(mysql_url)
        self.dbpool = adbapi.ConnectionPool('MySQLdb',
                                            charset='utf8',
                                            use_unicode=True,
                                            connect_timeout=5,
                                            **conn_kwargs)
    def close_spider(self, spider):
        self.dbpool.close()
    @defer.inlineCallbacks
    def process_item(self, item, spider):
        try:
            yield self.dbpool.runInteraction(self.do_replace, item)
        except:
            print traceback.format_exc()
        defer.returnValue(item)
    @staticmethod
    def do_replace(tx, item):
        sql = """REPLACE INTO properties (url, title, price, description) VALUES (%s,%s,%s,%s)"""
        args = (
            item["url"][0][:100],
            item["title"][0][:30],
            item["price"][0],
            item["description"][0].replace("\r\n", " ")[:30]
        )
        tx.execute(sql, args)

提示:完整代码位于ch09/properties/properties/pipelines/mysql.py。

本质上,这段代码的大部分都很普通。为了简洁而省略的代码将一条保存在MYSQL_PIPELINE_URL、格式是mysql://user:pass@ip/database的URL,解析成了独立的参数。在爬虫的init()中,将它们传递到adbapi.ConnectionPool(),它使用adbapi的底层结构,初始化MySQL连接池。第一个参数是我们想要引入的模块的名字。对于我们的MySQL,它是MySQLdb。我们为MySQL客户端另设了几个参数,以便正确处理Unicode和超时。每当adbapi打开新连接时,所有这些参数都要进入底层的MySQLdb.connect()函数。爬虫关闭时,我们调用连接池的close()方法。

我们的process_item()方法包装了dbpool.runInteraction()。这个方法给调回方法排队,会在当连接池中一个连接的Transaction对象变为可用时被调用。这个Transaction对象有一个和DB-API指针相似的API。在我们的例子中,调回方法是do_replace(),它定义在后面几行。@staticmethod是说这个方法关联的是类而不是具体的类实例,因此,我们可以忽略通常的self参数。如果方法不使用成员的话,最好设其为静态,如果你忘了设为静态也不要紧。这个方法准备了一个SQL字符串、几个参数,并调用Transaction的execute()函数,以进行插入。我们的SQL使用REPLACE INTO,而不用更常见的INSERT INTO,来替换键相同的项。这可以让我们的案例简化。如果我们相拥SQL返回数据,例如SELECT声明,我们使用dbpool.runQuery(),我们可能还需要改变默认指针,方法是设置adbapi.ConnectionPool()的参数cursorclass为cursorclass=MySQLdb.cursors,这样取回数据更为简便。

使用这个pipeline,我们要在settings.py的ITEM_PIPELINES添加它,还要设置一下MYSQL_PIPELINE_URL:

ITEM_PIPELINES = { ...
    'properties.pipelines.mysql.MysqlWriter': 700,
...
MYSQL_PIPELINE_URL = 'mysql://root:pass@mysql/properties'

执行以下命令:

scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=1000

运行这条命令后,返回MySQL控制台,可以看到如下记录:

mysql> SELECT COUNT(*) FROM properties;
+----------+
|     1006 |
+----------+
mysql> SELECT * FROM properties LIMIT 4;
+------------------+--------------------------+--------+-----------+
| url              | title                    | price  | description
+------------------+--------------------------+--------+-----------+
| http://...0.html | Set Unique Family Well   | 334.39 | website c
| http://...1.html | Belsize Marylebone Shopp | 388.03 | features                       
| http://...2.html | Bathroom Fully Jubilee S | 365.85 | vibrant own
| http://...3.html | Residential Brentford Ot | 238.71 | go court
+------------------+--------------------------+--------+-----------+
4 rows in set (0.00 sec)

延迟和吞吐量的性能和之前相同。结果让人印象深刻。

使用Twisted 特定客户端连接服务

目前为止,我们学习了如何用treq使用类REST APIs。Scrapy可以用Twisted特定客户端连接许多其它服务。例如,如果我们想连接MongoDB,通过搜索“MongoDB Python”,我们可以找到PyMongo,它是阻塞/同步的,除非我们使用pipeline处理阻塞操作中的线程,我们不能在Twisted中使用PyMongo。如果我们搜索“MongoDB Twisted Python”,可以找到txmongo,它可以完美适用于Twisted和Scrapy。通常的,Twisted客户端群体很小,但使用它比起自己写一个客户端还是要方便。下面,我们就使用这样一个Twisted特定客户端连接Redis键值对存储。

用pipeline读写Redis

Google Geocoding API是按照每个IP进行限制的。如果可以接入多个IPs(例如,多台服务器),当一个地址已经被另一台机器做过地理编码,就要设法避免对发出重复的请求。如果一个地址之前已经被查阅过,也要避免再次查阅。我们不想浪费限制的额度。

笔记:与API商家联系,以确保这符合规定。你可能,必须每几分钟/小时,就要清空缓存记录,或者根本就不能缓存。

我们可以使用Redis键值缓存作为分布式dict。Vagrant环境中已经有了一个Redis实例,我们现在可以连接它,用redis-cli作一些基本操作:

$ redis-cli -h redis
redis:6379> info keyspace
# Keyspace
redis:6379> set key value
OK
redis:6379> info keyspace
# Keyspace
db0:keys=1,expires=0,avg_ttl=0
redis:6379> FLUSHALL
OK
redis:6379> info keyspace
# Keyspace
redis:6379> exit

通过搜索“Redis Twisted”,我们找到一个txredisapi库。它最大的不同是,它不仅是一个Python的同步封装,还是一个Twisted库,可以通过reactor.connectTCP(),执行Twisted协议,连接Redis。其它库也有类似用法,但是txredisapi对于Twisted效率更高。我们可以通过安装库dj_redis_url可以安装它,这个库通过pip可以解析Redis配置URL(sudo pip install txredisapi dj_redis_url)。和以前一样,开发机中已经安装好了。

我们如下启动RedisCache pipeline:

from txredisapi import lazyConnectionPool
class RedisCache(object):
...
    def __init__(self, crawler, redis_url, redis_nm):
        self.redis_url = redis_url
        self.redis_nm = redis_nm
        args = RedisCache.parse_redis_url(redis_url)
        self.connection = lazyConnectionPool(connectTimeout=5,
                                             replyTimeout=5,
                                             **args)
        crawler.signals.connect(
                self.item_scraped,signal=signals.item_scraped)

这个pipeline比较简单。为了连接Redis服务器,我们需要主机、端口等等,它们全都用URL格式存储。我们用parse_redis_url()方法解析这个格式。使用命名空间做键的前缀很普遍,在我们的例子中,我们存储在redis_nm。我们然后使用txredisapi的lazyConnectionPool()打开一个数据库连接。

最后一行有一个有趣的函数。我们是想用pipeline封装geo-pipeline。如果在Redis中没有某个值,我们不会设定这个值,geo-pipeline会用API像之前一样将地址进行地理编码。完毕之后,我们必须要在Redis中缓存键值对,我们是通过连接signals.item_scraped信号来做的。我们定义的调回(即item_scraped()方法,马上会讲)只有在最后才会被调用,那时,地址就设置好了。

提示:完整代码位于ch09/properties/properties/pipelines/redis.py。

我们简化缓存,只寻找和存储每个Item的地址和地点。这对Redis来说是合理的,因为它通常是运行在单一服务器上的,这可以让它很快。如果不是这样的话,可以加入一个dict结构的缓存,它与我们在geo-pipeline中用到的相似。以下是我们如何处理入库的Items:

process incoming Items:
@defer.inlineCallbacks
def process_item(self, item, spider):
    address = item["address"][0]
    key = self.redis_nm + ":" + address
    value = yield self.connection.get(key)
    if value:
        item["location"] = json.loads(value)
    defer.returnValue(item)

和预期的相同。我们得到了地址,给它添加前缀,然后使用txredisapi connection的get()在Redis进行查找。我们将JSON编码的对象在Redis中保存成值。如果一个值设定了,我们就使用JSON解码,然后将其设为地点。

当一个Item到达pipelines的末端时,我们重新取得它,将其保存为Redis中的地点值。以下是我们的做法:

    from txredisapi import ConnectionError
    def item_scraped(self, item, spider):
        try:
            location = item["location"]
            value = json.dumps(location, ensure_ascii=False)
        except KeyError:
            return
        address = item["address"][0]
        key = self.redis_nm + ":" + address
        quiet = lambda failure: failure.trap(ConnectionError)
        return self.connection.set(key, value).addErrback(quiet)

如果我们找到了一个地点,我们就取得了地址,添加前缀,然后使用它作为txredisapi连接的set()方法的键值对。set()方法没有使用@defer.inlineCallbacks,因为处理signals.item_scraped时,它不被支持。这意味着,我们不能对connection.set()使用yield,但是我们可以返回一个延迟项,Scrapy可以在它后面排上其它信号对象。任何情况下,如果Redis的连接不能使用用connection.set(),它就会抛出一个例外。在这个错误处理中,我们把传递的错误当做参数,我们让它trap()任何ConnectionError。这是Twisted的延迟API的优点之一。通过用trap()捕获错误项,我们可以轻易忽略它们。

使这个pipeline生效,我们要做的是在settings.py的ITEM_PIPELINES中添加它,并提供一个REDIS_PIPELINE_URL。必须要让它的优先级比geo-pipeline高,以免太晚就不能使用了:

ITEM_PIPELINES = { ...
    'properties.pipelines.redis.RedisCache': 300,
    'properties.pipelines.geo.GeoPipeline': 400,
...
REDIS_PIPELINE_URL = 'redis://redis:6379'

像之前一样运行。第一次运行时和以前很像,但随后的运行结果如下:

$ scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=100
...
INFO: Enabled item pipelines: TidyUp, RedisCache, GeoPipeline, 
MysqlWriter, EsWriter
...
Scraped... 0.0 items/s, avg latency: 0.00 s, time in pipelines: 0.00 s
Scraped... 21.2 items/s, avg latency: 0.78 s, time in pipelines: 0.15 s
Scraped... 24.2 items/s, avg latency: 0.82 s, time in pipelines: 0.16 s
...
INFO: Dumping Scrapy stats: {...
   'geo_pipeline/already_set': 106,
   'item_scraped_count': 106,

我们看到GeoPipeline和RedisCache都生效了,RedisCache第一个输出。还注意到在统计中geo_pipeline/already_set: 106。这是GeoPipeline发现的Redis缓存中填充的数目,它不调用Google API。如果Redis缓存是空的,你会看到Google API处理了一些键。从性能上来讲,我们看到GeoPipeline引发的初始行为消失了。事实上,当我们开始使用内存,我们绕过了每秒只有5次请求的API限制。如果我们使用Redis,应该考虑使用过期键,让系统周期刷新缓存数据。

连接CPU密集型、阻塞或旧方法

最后一部分讲连接非Twisted的工作。尽管异步程序的优点很多,并不是所有库都专门为Twisted和Scrapy写的。使用Twisted的线程池和reactor.spawnProcess()方法,我们可以使用任何Python库和任何语言写的编码。

pipeline进行CPU密集型和阻塞操作

我们在第8章中强调,反应器适合简短非阻塞的任务。如果我们不得不要处理复杂和阻塞的任务,又该怎么做呢?Twisted提供了线程池,有了它可以使用reactor.callInThread() API在分线程而不是主线程中执行慢操作。这意味着,反应器可以一直运行并对事件反馈,而不中断计算。但要记住,在线程池中运行并不安全,当你使用全局模式时,会有多线程的同步问题。让我们从一个简单的pipeline开始,逐渐做出完整的代码:

class UsingBlocking(object):
    @defer.inlineCallbacks
    def process_item(self, item, spider):
        price = item["price"][0]
        out = defer.Deferred()
        reactor.callInThread(self._do_calculation, price, out)
    item["price"][0] = yield out
        defer.returnValue(item)
    def _do_calculation(self, price, out):
        new_price = price + 1
        time.sleep(0.10)
        reactor.callFromThread(out.callback, new_price)

在前面的pipeline中,我们看到了一些基本用法。对于每个Item,我们提取出价格,我们相用_do_calucation()方法处理它。这个方法使用time.sleep(),一个阻塞操作。我们用reactor.callInThread()调用,让它在另一个线程中运行。显然,我们传递价格,我们还创建和传递了一个名为out的延迟项。当_do_calucation()完成了计算,我们使用out调回值。下一步,我们执行延迟项,并未价格设新的值,最后返回Item。

在_do_calucation()中,有一个细微之处,价格增加了1,进而睡了100ms。这个时间很多,如果调用进反应器主线程,每秒就不能抓取10页了。通过在另一个线程中运行,就不会再有这个问题。任务会在线程池中排队,每次处理耗时100ms。最后一步是触发调回。一般的,我们可以使用out.callback(new_price),但是因为我们现在是在另一个线程,这么做不安全。如果这么做的话,延迟项的代码会被从另一个线程调用,这样迟早会产生错误的数据。不这样做,转而使用reactor.callFromThread(),它也可以将函数当做参数,将任意其余参数传递到函数。这个函数会排队并被调回主线程,主进程反过来会打开process_item()对象yield,并继续Item的操作。

如果我们用全局模式,例如计数器、滑动平均,又该怎么使用_do_calucation()呢?例如,添加两个变量,beta和delta,如下所示:

class UsingBlocking(object):
    def __init__(self):
        self.beta, self.delta = 0, 0
    ...
    def _do_calculation(self, price, out):
        self.beta += 1
        time.sleep(0.001)
self.delta += 1
        new_price = price + self.beta - self.delta + 1
        assert abs(new_price-price-1) < 0.01
        time.sleep(0.10)...

这段代码是断言失败错误。这是因为如果一个线程在self.beta和self.delta间切换,另一个线程继续计算使用beta/delta计算价格,它会发现它们状态不一致(beta大于delta),因此,计算出错误的结果。短暂的睡眠可能会造成竞争条件。为了不发生这些状况,我们要使一个锁,例如Python的threading.RLock()递归锁。使用它,可以确保没有两个线程在同一时间操作被保护代码:

class UsingBlocking(object):
    def __init__(self):
        ...
        self.lock = threading.RLock()
    ...
    def _do_calculation(self, price, out):
        with self.lock:
            self.beta += 1
            ...
            new_price = price + self.beta - self.delta + 1
        assert abs(new_price-price-1) < 0.01 ...

代码现在就正确了。记住,我们不需要保护整段代码,就足以处理全局模式。

提示:完整代码位于ch09/properties/properties/pipelines/computation.py。

要使用这个pipeline,我们需要把它添加到settings.py的ITEM_PIPELINES中。如下所示:

ITEM_PIPELINES = { ...
    'properties.pipelines.computation.UsingBlocking': 500,

像之前一样运行爬虫,pipeline延迟达到了100ms,但吞吐量没有发生变化,大概每秒25个items。

pipeline使用二进制和脚本

最麻烦的借口当属独立可执行文件和脚本。打开需要几秒(例如,从数据库加载数据),但是后面处理数值的延迟很小。即便是这种情况,Twisted也预料到了。我们可以使用reactor.spawnProcess() API和相关的protocol.ProcessProtocol来运行任何执行文件。让我们来看一个例子。脚本如下:

#!/bin/bash
trap "" SIGINT
sleep 3
while read line
do
    # 4 per second
    sleep 0.25
    awk "BEGIN {print 1.20 * $line}"
done

这是一个简单的bash脚本。它运行时,会使Ctrl + C 无效。这是为了避免系统的一个奇怪的错误,将Ctrl + C增值到子流程并过早结束,导致Scrapy强制等待流程结果。在使Ctrl + C无效之后,它睡眠三秒,模拟启动时间。然后,它阅读输入的代码语句,等待250ms,然后返回结果价格,价格的值乘以了1.20,由Linux的awk命令计算而得。这段脚本的最大吞吐量为每秒1/250ms=4个Items。用一个短session检测:

$ properties/pipelines/legacy.sh 
12 <- If you type this quickly you will wait ~3 seconds to get results
14.40
13 <- For further numbers you will notice just a slight delay
15.60

因为Ctrl + C失效了,我们用Ctrl + D必须结束session。我们该如何让Scrapy使用这个脚本呢?再一次,我们从一个简化版开始:

class CommandSlot(protocol.ProcessProtocol):
    def __init__(self, args):
      self._queue = []
        reactor.spawnProcess(self, args[0], args)
    def legacy_calculate(self, price):
        d = defer.Deferred()
        self._queue.append(d)
        self.transport.write("%f\n" % price)
        return d
    # Overriding from protocol.ProcessProtocol
    def outReceived(self, data):
        """Called when new output is received"""
        self._queue.pop(0).callback(float(data))
class Pricing(object):
    def __init__(self):
        self.slot = CommandSlot(['properties/pipelines/legacy.sh'])
    @defer.inlineCallbacks
    def process_item(self, item, spider):
        item["price"][0] = yield self.slot.legacy_calculate(item["price"][0])
       defer.returnValue(item)

我们在这里找到了一个名为CommandSlot的ProcessProtocol和Pricing爬虫。在init()中,我们创建了新的CommandSlot,它新建了一个空的队列,并用reactor.spawnProcess()开启了一个新进程。它调用收发数据的ProcessProtocol作为第一个参数。在这个例子中,是self的原因是spawnProcess()是被从类protocol调用的。第二个参数是可执行文件的名字,第三个参数args,让二进制命令行参数成为字符串序列。

在pipeline的process_item()中,我们用CommandSlot的legacy_calculate()方法代表所有工作,CommandSlot可以返回产生的延迟项。legacy_calculate()创建延迟项,将其排队,用transport.write()将价格写入进程。ProcessProtocol提供了transport,可以让我们与进程沟通。无论何时我们从进程收到数据, outReceived()就会被调用。通过延迟项,进程依次执行,我们可以弹出最老的延迟项,用收到的值触发它。全过程就是这样。我们可以让这个pipeline生效,通过将它添加到ITEM_PIPELINES:

ITEM_PIPELINES = {...
    'properties.pipelines.legacy.Pricing': 600,

如果运行的话,我们会看到性能很差。进程变成了瓶颈,限制了吞吐量。为了提高性能,我们需要修改pipeline,允许多个进程并行运行,如下所示:

class Pricing(object):
    def __init__(self):
        self.concurrency = 16
        args = ['properties/pipelines/legacy.sh']
        self.slots = [CommandSlot(args) 
                      for i in xrange(self.concurrency)]
        self.rr = 0
    @defer.inlineCallbacks
    def process_item(self, item, spider):
        slot = self.slots[self.rr]
        self.rr = (self.rr + 1) % self.concurrency
        item["price"][0] = yield
                         slot.legacy_calculate(item["price"][0])
        defer.returnValue(item)

这无非是开启16个实例,将价格以轮转的方式发出。这个pipeline的吞吐量是每秒16*4 = 64。我们可以用下面的爬虫进行验证:

 $ scrapy crawl easy -s CLOSESPIDER_ITEMCOUNT=1000
...
Scraped... 0.0 items/s, avg latency: 0.00 s and avg time in pipelines: 
0.00 s
Scraped... 21.0 items/s, avg latency: 2.20 s and avg time in pipelines: 
1.48 s
Scraped... 24.2 items/s, avg latency: 1.16 s and avg time in pipelines: 
0.52 s

延迟增加了250 ms,但吞吐量仍然是每秒25。
请记住前面的方法使用了transport.write()让所有的价格在脚本shell中排队。这个可能对你的应用不适用,,尤其是当数据量很大时。Git的完整代码让值和调回都进行了排队,不想脚本发送值,除非收到前一项的结果。这种方法可能看起来更友好,但是会增加代码复杂度。

总结

你刚刚学习了复杂的Scrapy pipelines。目前为止,你应该就掌握了所有和Twisted编程相关的知识。并且你学会了如何在进程中执行复杂的功能,用Item Processing Pipelines存储Items。我们看到了添加pipelines对延迟和吞吐量的影响。通常,延迟和吞吐量是成反比的。但是,这是在恒定并发数的前提下(例如,一定数量的线程)。在我们的例子中,我们一开始的并发数为N=ST=250.77≅19,添加pipelines之后,并发数为N=25*3.33≅83,并没有引起性能的变化。这就是Twisted的强大之处!下面学习第10章,Scrapy的性能。


序言
第1章 Scrapy介绍
第2章 理解HTML和XPath
第3章 爬虫基础
第4章 从Scrapy到移动应用
第5章 快速构建爬虫
第6章 Scrapinghub部署
第7章 配置和管理
第8章 Scrapy编程
第9章 使用Pipeline
第10章 理解Scrapy的性能
第11章(完) Scrapyd分布式抓取和实时分析


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
Py之imblearn:imblearn/imbalanced-learn库的简介、安装、使用方法之详细攻略
Py之imblearn:imblearn/imbalanced-learn库的简介、安装、使用方法之详细攻略
328 0
使用ping6检查LINUX服务器是否支持IPV6
苹果上架,现在IPV6是必备条件。前几天为其他公司服务器出的版本被拒之后,吾决定先用吾司服务器上架再说。感觉吾司服务器是支持IPV6的。 那么,吾司服务器是不是真的支持IPV6?于是上网查了一下,边学习边记录,希望给后来的朋友带来方便:
36 0
deep learning的使用观点
为保证作者原意,不做翻译 cited from: Adrian Rosebrock Get off the deep learning bandwagon and get some perspective Disclaimer: This post is a bit cynical in tone.
1273 0
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
20690 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
18988 0
机器学习之使用sklearn构造决策树模型
机器学习之使用sklearn构造决策树模型一、任务基础导入所需要的库 1234import matplotlib.pyplot as pltimport pandas as pd %matplotlib inline加载sklearn内置数据集 ,查看数据描述 123from sklearn.
749 0
Xamarin android spinner的使用方法
xamarin  android spinner的如何使用呢,大多数web开发人员经常会听到DropDownList 和Combobox 这种下拉选择框,spinner 的意思差不多,有道词典一下意思是“下拉列表”、“台湾斯普”,“下拉列表组件”,“微调控件”。
1213 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
25226 0
使用Q-learning算法让Chrome小恐龙自动学习躲避障碍物(示例)
原理很简单,就是给定小恐龙障碍物类型、障碍物距离、游戏的相对速度作为状态,小恐龙有跳跃,低头,不做任何动作这三种动作可选,小恐龙存活则奖励碰到障碍物则惩罚。
1916 0
+关注
seancheney
热 &middot; 爱 &gt;&gt; 最近在看的书: 《Pandas Cookbook》 《图解数据结构&middot;使用Python》 《利用Apache Flink进行流处理》 &gt;&gt; 欢迎指出译稿和其它文章中的错误 &gt;&gt; 所有文章仅供个人学习不涉及商业利益
117
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
JS零基础入门教程(上册)
立即下载
性能优化方法论
立即下载
手把手学习日志服务SLS,云启实验室实战指南
立即下载