[Python] AWSのサーバーレスアーキテクチャを使って、イベントドリブンなWebクローラーを作ってみる


# サーバーレスでイベントドリブンなWEBクローラーを作ってみる
今ElasticSearchを勉強中なので、ESを使って何かできないかなと思い、Kinesis+Lambdaを使ったイベントドリブンなWebクローラーを書いてみました。

- 実行環境
    - CentOS7
    - python 2.7

## ワークフロー
![serverless-crawler.png](https://qiita-image-store.s3.amazonaws.com/0/121487/ca7cf447-a867-dd71-2810-7e72a7c5037e.png)

大まかな流れは以下の通りです。
1. Scrapy(ScrapingHub or AWS Lambda)でURLを抽出、KinesisストリームにPUT
2. KinesisストリームからAWS Lambdaに発火させる
3. Lambda関数からURLへ巡回し、データをElasticSearch Serviceへ流す

## IAMユーザーの作成
KinesisとElasticSearchを使う上で、権限が必要になるので、
それぞれに**アクセスキーID**と**シークレットアクセスキー**を用意しておく。

また、ユーザーのARNも必要になるのでひかえておく(arn:aws:iam::＊＊＊＊＊＊＊＊＊＊:user/＊＊＊＊＊＊＊)

## AWS Kinesisでストリームを作成する
まずはKinesisストリームを作成します。
1. ストリームを作成
2. 適当なストリーム名を入力 (仮にscraping_url)
3. シャード数を入力 (とりあえず1)
4. ストリームの作成
![スクリーンショット 2017-04-06 12.45.52.png](https://qiita-image-store.s3.amazonaws.com/0/121487/9af3d859-410c-c6ef-1e9f-56a02a105c39.png)

## AWS ElasticSearch Serviceを作成する
次にAmazon ElasticSearch ServiceでESを作成します。
### AWSでの操作
1. Create a new domain
2. Elasticsearch domain nameに適当なドメイン名を入力 (仮にweb-archives)
3. Elasticsearch versionは[5.1]を選択。[Next]を押す
![スクリーンショット 2017-04-06 13.06.02.png](https://qiita-image-store.s3.amazonaws.com/0/121487/100056a1-170a-9c72-359e-8aa951de0158.png)
4. Configure clusterでInstance typeを[t2.small]にしておく。[Next]押す (テストなので小さいインスタンスで)
![スクリーンショット 2017-04-06 13.07.46.png](https://qiita-image-store.s3.amazonaws.com/0/121487/b0703f09-f45c-b161-f3e0-705d452b3067.png)

5. Set up access policyで[Allow or deny access to one or more AWS accounts or IAM users]を選択
![スクリーンショット 2017-04-06 13.18.13.png](https://qiita-image-store.s3.amazonaws.com/0/121487/d41f7bcd-2b49-ea6a-a75f-ca0b7adc7843.png)

6. Account ID or ARN*に許可したいユーザーのARNを入力
![スクリーンショット 2017-04-06 13.18.51.png](https://qiita-image-store.s3.amazonaws.com/0/121487/5e0a989c-4e65-f35c-d3c9-b1899230d6d1.png)

7. [Confirm and create]で作成
8. しばらくするとESが立ち上がるので[Endpoint]を確認し、Lambdaで使うのでひかえておく
![スクリーンショット 2017-04-06 13.30.18.png](https://qiita-image-store.s3.amazonaws.com/0/121487/2d604fdd-4125-74a5-5969-9e2893467b00.png)


### ElasticSearchにインデックス作成とマッピングをする
記事保存用にURL,タイトル,記事内容を保存するマッピングデータを作成します。
```mapping.json
{
    "mappings": {                                            
        "article": {
            "properties" : {
                "url" : {
                    "type": "string",
                    "index" : "not_analyzed"
                },
                "title" : {
                    "type": "string",
                    "index" : "analyzed"
                },
                "contents" : {
                    "type": "string",
                    "index" : "analyzed"
                }
            }
        }   
    }
}
```
次に上記マッピングデータと、インデックス作成をするスクリプトを作成します。

予めローカルに以下のパッケージをインストールしておきます<br>
`$ pip install requests_aws4auth elasticsearch`

```es-mapping.py
# -*- coding: utf-8 -*-
import elasticsearch
from requests_aws4auth import AWS4Auth
import json

if __name__ == '__main__':
    # ESのエンドポイントを指定
    host='search-***************.ap-northeast-1.es.amazonaws.com'
    awsauth = AWS4Auth(
            # AWSユーザーのアクセスキーIDとシークレットアクセスキー
            'ACCESS_KRY_ID',
            'SECRET_ACCESS_KEY',
            'ap-northeast-1', 'es')

    es = elasticsearch.Elasticsearch(
            hosts=[{'host': host, 'port': 443}],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=elasticsearch.connection.RequestsHttpConnection
            )

    f = open('mapping.json', 'r')
    mapping = json.load(f)

    es.indices.create(index='website')                    
    es.indices.put_mapping(index='website', doc_type='article', body=mapping['mappings'])
```
`$ python es-mapping.py`<br>
スクリプトを実行すると、AWS ES上にインデックスが作成されているはずです。
![スクリーンショット 2017-04-06 15.32.52.png](https://qiita-image-store.s3.amazonaws.com/0/121487/38eb98c5-a62a-9e81-cb70-ef6d184c1eb6.png)

## AWS Lambdaを作成する
ElasticSearchを作成したので、次にLambda Functionを作成します。

### Lambda関数の作成
ローカルにLambda関数を作成します。
`$ mkdir web_crawler`<br>
`$ cd web_crawler`<br>
`$ vim lambda_function.py`

```lambda_function.py

# -*- coding: utf-8 -*-                    
import os
import base64
from readability import Document
import html2text
import requests
import elasticsearch
from elasticsearch import helpers
from requests_aws4auth import AWS4Auth

def lambda_handler(event, context):
    host = os.environ['ES_HOST']
    # ElasticSearch Serviceへの認証にIAM Roleを利用する
    awsauth = AWS4Auth(
            os.environ['ACCESS_ID'],
            os.environ['SECRET_KEY'], 'ap-northeast-1', 'es')

    es = elasticsearch.Elasticsearch(
            hosts=[{'host': host, 'port': 443}],
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=elasticsearch.connection.RequestsHttpConnection
    )

    articles = []

    # Kinesis Streamからイベントを取得
    for record in event['Records']:
        payload = base64.b64decode(record['kinesis']['data'])
        try:
            response = requests.get(payload)
            if response.ok:
                article = Document(response.content).summary()
                titleText = html2text.html2text(Document(response.content).title())
                contentsText = html2text.html2text(article)
                res = es.search(index="website", body={"query": {"match": {"url": payload}}})
                # ESにURLが既に登録されているか
                if res['hits']['total'] is 0:
                    doc = {
                        'url': payload,
                        'title': titleText.encode('utf-8'),
                        'contents': contentsText.encode('utf-8')
                    }
                    articles.append({'_index':'website', '_type':'scraper', '_source':doc})
        except requests.exceptions.HTTPError as err:
            print("HTTPError: " + err)                                                                                                                                                       
    # Bulk Insert
    helpers.bulk(es, articles)
```
Lambda関数を作成したら、必要なライブラリを同一階層へインストール<br>
`$ pip install readability-lxml html2text elasticsearch requests_aws4auth requests -t /path/to/web_crawler`<br>
zipで固めます<br>
`$ zip -r web_crawler.zip .`

### AWSにLambda関数をデプロイする
1. [Lamnda 関数の作成]
1. [ブランク関数]を選択
1. [トリガーの設定]で先ほど作成したKinesisストリームを選択する
  1. [バッチサイズ]は10程度
  1. [開始位置]は水平トリム
  1. トリガーの有効化をチェック
![スクリーンショット 2017-04-06 13.51.27.png](https://qiita-image-store.s3.amazonaws.com/0/121487/10f90735-4211-9270-1919-bb50c151b6e0.png)

1. [関数の設定]で[名前]を入力する（ここでは仮にWebCrawler）
  1. [ランタイム]はPython 2.7を選択
  1. [コード エントリ タイプ]で.ZIPファイルをアップロードを選択
  1. [関数パッケージ]から先ほど作成したzipファイルを指定
  1.  [環境変数]にElasticSearchへのアクセス用に３つ設定する
    1. ACCESS_IDにアクセスキーID
    1. SECRET_KEYにシークレットアクセスキー
    1. ES_HOSTにElasticSearchのエンドポイント
![スクリーンショット 2017-04-06 16.24.30.png](https://qiita-image-store.s3.amazonaws.com/0/121487/7be93c54-76d2-1268-7214-280b02bed810.png)

  1. [ハンドラ]はlambda_function.lambda_handlerのまま
  1. ロールは適宜作成
  1. [詳細設定]の[タイムアウト]を2分くらいにする
  1. [関数の作成]

## ScrapyでURL抽出、Kinesisストリームへ送信する
次は最終段階、Scrapyを使って一覧ページからURLを抽出し、Kinesisストリームへデータを送信してみます。

一覧ページは、はてなブックマークのホットエントリーを使います。Scrapyを使えばRSSの方が簡単にデータを取れそうなのですが、あえてWebページからスクレイピングしてみました。Scrapyは高度なWebクローラーを作る際、便利で強力なフレームワークなので、興味があれば触ってみてください。

* [Scrapy](https://scrapy.org/)

### プロジェクトの作成
まずはScrapyをインストール<br>
`$ pip install scrapy`<br>
`$ scrapy startproject hotentry`<br>
`$ vim hotentry/hotentry/spiders/hotentry.py`<br>
以下のコードを入力します。

```hotentry.py
# -*- coding: utf-8 -*-
import scrapy
from scrapy.conf import settings
import boto3
import json

kinesis = boto3.client(
        'kinesis',                                                                                                                                                                           
        aws_access_key_id=settings['AWS_ACCESS_KEY_ID'],
        aws_secret_access_key=settings['AWS_SECRET_ACCESS_KEY'],
        region_name='ap-northeast-1')

class HotEntrySpider(scrapy.Spider):
    name = "hotentry"
    allowed_domains = ["b.hatena.ne.jp"]
    start_urls = ['http://b.hatena.ne.jp/hotentry/general']

    def parse(self, response):
        for sel in response.css("li.hb-entry-unit-with-favorites"):
            url = sel.css("a.entry-link::attr('href')").extract_first()
            if url is None:
                continue
            kinesis.put_record(
                    StreamName = "scraping_url",
                    Data = sel.css("a.entry-link::attr('href')").extract_first(),
                    PartitionKey = "scraper"
            )
```
`$ vim hotentry/hotentry/settings.py`<br>
settings.pyにアクセスキーIDとシークレットアクセスキーを追記
```
AWS_ACCESS_KEY_ID = 'AKI******************'
AWS_SECRET_ACCESS_KEY = '************************************'
```

これでKinesisストリームへPUTすることができるようになりました。試しにこのコードを実行してみます。<br>
`$ scrapy crawl hotenty`<br>

これで「Scrapy -> Kinesis -> AWS Lambda -> ElasticSearch」とデータが投入できたはずです。

### ScrapyをScrapinghubへデプロイする
ScrapyでURLを抽出し、Kinesisへ送る事ができましたが、このままではローカルバッチになってしまいますので、ScrapyのコードをScrapinghubというクラウドサービスへデプロイします。

導入方法は以下の記事が詳しいので、こちらを見てください。<br>
* [Scrapy + Scrapy Cloudで快適Pythonクロール+スクレイピングライフを送る](http://data.gunosy.io/entry/python-scrapy-scraping)

ユーザー登録からデプロイまでは簡単に出来てしまうので端折ります。

## 最後に
当初はSQSやDynamoDB使いLambda関数も複数にわけていたのですが、複雑になってエラーが追えず挫折しました。やはりシンプルイズベスト。Lambdaのトリガーはもっと対応サービス増やしてほしいなぁ。 

**※ このコードはテストで書いたものなので、エラーハンドリングなどを厳密に行っていません。このコードで何かしら不利益があったとしても自己責任でお願いします。**