作者：俊欣

來源：關于數據分析與可視化

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

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

發起請求

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

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)

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

響應內容解析

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

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 

最后則是將解析出來的內容存入至數據庫當中

數據存儲

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

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))

我們通過調用ensure_future方法來安排協程的進行

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

數據分析與可視化

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

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="房屋面積大小的區間", 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

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

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

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