作者：俊欣

來源：關于數(shù)據(jù)分析與可視化

今天我們來聊一下如何用協(xié)程來進行數(shù)據(jù)的抓取，協(xié)程又稱為是微線程，也被稱為是用戶級線程，在單線程的情況下完成多任務，多個任務按照一定順序交替執(zhí)行。

那么aiohttp模塊在Python中作為異步的HTTP客戶端/服務端框架，是基于asyncio的異步模塊，可以用于實現(xiàn)異步爬蟲，更快于requests的同步爬蟲。下面我們就通過一個具體的案例來看一下該模塊到底是如何實現(xiàn)異步爬蟲的。

發(fā)起請求

我們先來看一下發(fā)起請求的部分，代碼如下

async def fetch(url, session): try: async with session.get(url, headers=headers, verify_ssl=False) as resp: if resp.status in [200, 201]:
                logger.info("請求成功")
                data = await resp.text() return data except Exception as e:
        print(e)
        logger.warning(e)

要是返回的狀態(tài)碼是200或者是201，則獲取響應內(nèi)容，下一步我們便是對響應內(nèi)容的解析

響應內(nèi)容解析

這里用到的是PyQuery模塊來對響應的內(nèi)容進行解析，代碼如下

def extract_elements(source): try:
        dom = etree.HTML(source)
        id = dom.xpath('......')[0]
        title = dom.xpath('......')[0]
        price = dom.xpath('.......')[0]
        information = dict(re.compile('.......').findall(source))
        information.update(title=title, price=price, url=id)
        print(information)
        asyncio.ensure_future(save_to_database(information, pool=pool)) except Exception as e:
        print('解析詳情頁出錯！')
        logger.warning('解析詳情頁出錯！') pass 

最后則是將解析出來的內(nèi)容存入至數(shù)據(jù)庫當中

數(shù)據(jù)存儲

這里用到的是aiomysql模塊，使用異步IO的方式保存數(shù)據(jù)到Mysql當中，要是不存在對應的數(shù)據(jù)表，我們則創(chuàng)建對應的表格，代碼如下

async def save_to_database(information, pool): COLstr = '' # 列的字段 ROWstr = '' # 行字段 ColumnStyle = ' VARCHAR(255)' if len(information.keys()) == 14: for key in information.keys():
            COLstr = COLstr + ' ' + key + ColumnStyle + ',' ROWstr = (ROWstr + '"%s"' + ',') % (information[key]) async with pool.acquire() as conn: async with conn.cursor() as cur: try: await cur.execute("SELECT * FROM  %s" % (TABLE_NAME)) await cur.execute("INSERT INTO %s VALUES (%s)" % (TABLE_NAME, ROWstr[:-1])) except aiomysql.Error as e: await cur.execute("CREATE TABLE %s (%s)" % (TABLE_NAME, COLstr[:-1])) await cur.execute("INSERT INTO %s VALUES (%s)" % (TABLE_NAME, ROWstr[:-1])) except aiomysql.Error as e: pass 

項目的啟動

最后我們來看一下項目啟動的代碼，如下

async def consumer():
    async with aiohttp.ClientSession() as session: while not stop: if len(urls) != 0:
                _url = urls.pop() source = await fetch(_url, session)
                extract_links(source) if len(links_detail) == 0:
                print('目前沒有待爬取的鏈接')
                await asyncio.sleep(np.random.randint(5, 10))
                continue link = links_detail.pop() if link not in crawled_links_detail:
                asyncio.ensure_future(handle_elements(link, session))

我們通過調(diào)用ensure_future方法來安排協(xié)程的進行

async def handle_elements(link, session): print('開始獲取: {}'.format(link))
    source = await fetch(link, session) # 添加到已爬取的集合中 crawled_links_detail.add(link)
    extract_elements(source)

數(shù)據(jù)分析與可視化

下面我們針對抓取到的數(shù)據(jù)進行進一步的分析與可視化，數(shù)據(jù)源是關于上海的二手房的相關信息，我們先來看一下房屋戶型的分布，代碼如下

house_size_dict = {}
for house_size, num in zip(df["房屋戶型"].value_counts().head(10).index, df["房屋戶型"].value_counts().head(10).tolist()):
    house_size_dict[house_size] = num

print(house_size_dict)

house_size_keys_list = [key for key, values in house_size_dict.items()]
house_size_values_list = [values for key, values in house_size_dict.items()]

p = (
    Pie(init_opts=opts.InitOpts(theme=ThemeType.LIGHT))
        .add("", [list(z) for z in zip(house_size_keys_list, house_size_values_list)],
             radius=["35%", "58%"],
             center=["58%", "42%"])
        .set_global_opts(title_opts=opts.TitleOpts(title="房屋面積大小的區(qū)間", pos_left="40%"),
                         legend_opts=opts.LegendOpts(orient="vertical",
                                                     pos_top="15%",
                                                     pos_left="10%"))
        .set_series_opts(label_opts=opts.LabelOpts(formatter=": {c}"))
)

p.render("house_size.html")

output

我們可以看到占到大多數(shù)的都是“2室1廳1廚1衛(wèi)”的戶型，其次便是“1室1廳1廚1衛(wèi)”的戶型，可見上海二手房交易的市場賣的小戶型為主。而他們的所在樓層，大多也是在高樓層(共6層)的為主，如下圖所示

我們再來看一下房屋的裝修情況，市場上的二手房大多都是以“簡裝”或者是“精裝”為主，很少會看到“毛坯”的存在，具體如下圖所示

至此，我們就暫時先說到這里，本篇文章主要是通過異步協(xié)程的方式來進行數(shù)據(jù)的抓取，相比較于常規(guī)的requests數(shù)據(jù)抓取而言，速度會更快一些。