PythonでTwitterの検索結果をDynamoDBに入れ続ける

Twitterのクローラを作ったので記事としてまとめてみました。
どなたかの参考になれば嬉しいです。
ご指摘・ご質問ございましたらコメントにてお願いします。

Twitter APIのためのOAuth認証キーの取得

OAuth認証のために次の4つが必要です。

• Consumer key
• Consumer secret
• Access token
• Access token secret

これらをTwitter Application Managementで取得します。
なお、Twitterで携帯電話による認証を行っていない方はアプリケーションの作成ができませんので、Twitterの設定から携帯電話による認証を行ってください。

OAuth認証キーを取得する手順は次のとおりです。

1. Twitter Application Managementにログイン後、「Create New App」をクリックする。
2. 「Name」、「Description」、「Website」を（適当に）入力する。
3. 「Developer Agreement」にチェックを入れて、「Create your Twitter application」をクリックする。
4. アプリケーションが作れたら「Keys and Access Tokens」で「Create my access token」をクリックする。（なぜかエラーが出るが問題ない）
5. 次の4つが確認できたら完了。
   • Consumer Key (API Key)
   • Consumer Secret (API Secret)
   • Access Token
   • Access Token Secret

Pythonの実行環境を用意

今回はAmazon EC2を使います。将来的にデータ分析を行うことを想定して、Deep Learning AMIを選びます。
ただし、Deep Learning AMIは最低50GBのストレージが必要ですので、課金が発生します。ご注意ください。

開発用にJupyter Notebookの起動

最終的には自動実行するためIPython Notebook(.ipynb)からPython(.py)に変換するのですが、最初に開発するときはJupyter Notebookを使います。
EC2にSSHで接続してCUIで開発するよりは格段に楽だと思います。
Deep Learning AMIを選んで入ればJupyter Notebookは標準でインストールされているはずです。

折角なのでEC2起動時にJupyter Notebookも起動するようにしましょう。
手順は次のとおりです。

1. Jupyter Notebookにパスワードを設定する。
2. localhost以外からのアクセスを許可する。
3. Jupyter Notebookを自動起動するように設定する。

Jupyter Notebookにパスワードを設定

Deep Learning AMIでは最初からjupyter_notebook_config.pyが作られているので、--generate-configは不要です。
次のコマンドでパスワードのハッシュ値を生成します。Enter password:とVerify password:で2回パスワードが求められます。

$ ipython

In [1]: from notebook.auth import passwd

In [2]: passwd()
Enter password: # パスワードを入力
Verify password: # 同じパスワードを入力
Out[2]: 'sha1:87d2ccf6b4c6:810f745d61c4885cb9c1639f6bf3a263f7093cbf'

In [3]: # Ctrl+D
Do you really want to exit ([y]/n)? y

生成されたハッシュ値をコピーして、jupyter_notebook_config.pyの218行目付近に次のように追記してください。

$ vi ~/.jupyter/jupyter_notebook_config.py

...

## Hashed password to use for web authentication.
#  
#  To generate, type in a python/IPython shell:
#  
#    from notebook.auth import passwd; passwd()
#  
#  The string should be of the form type:salt:hashed-password.
#c.NotebookApp.password = ''
c.NotebookApp.password = u'sha1:87d2ccf6b4c6:810f745d61c4885cb9c1639f6bf3a263f7093cbf'

...

localhost以外からのアクセスを許可

次のコマンドでSSL用の鍵と証明書を生成します。

$ openssl req -x509 -nodes -newkey rsa:2048 -keyout mycert.key -out mycert.pem
$ chmod 600 mycert.key
$ chmod 600 mycert.pem

生成できたら、jupyter_notebook_config.pyに次のように追記してください。

c.NotebookApp.certfile = u'/home/ubuntu/mycert.pem'
c.NotebookApp.keyfile = u'/home/ubuntu/mycert.key'

localhost以外からのアクセスを許可するために、次の設定も変えておきましょう。

c.NotebookApp.ip = '*'

ついでにその他の設定も変えておきましょう。

c.NotebookApp.notebook_dir = '/home/ubuntu/workspace' # ルートディレクトリ
c.NotebookApp.open_browser = False # 起動時にブラウザを立ち上げるか否か

Jupyter Notebookはデフォルトだと8888番ポートで立ち上がるので、EC2のセキュリティグループのインバウンドに次のルールを追加してください。

項目	値
タイプ	カスタム TCP ルール
プロトコル	TCP
ポート範囲	8888
ソース	0.0.0.0/0
説明	（任意）

Jupyter Notebookを自動起動するように設定

/etc/rc.localを次のように書き換えます。

$ sudo vi /etc/rc.local
#!/bin/sh -e
#
# rc.local
#
# This script is executed at the end of each multiuser runlevel.
# Make sure that the script will "exit 0" on success or any other
# value on error.
#
# In order to enable or disable this script just change the execution
# bits.
#
# By default this script does nothing.

sudo -u ubuntu jupyter notebook &
exit 0

書き換えたらEC2を再起動してJupyter Notebookが起動するか確認してみましょう。
Webブラウザで https://[EC2のIPアドレス]:8888 を表示してみてください。
オレオレ証明書なのでブラウザの警告が表示されますが、気にしなくても大丈夫です。

DynamoDBの用意

DynamoDBは作るだけなら名前とプライマリキーを決めてポチポチするだけで作れます。
AWSコンソールを開いて、DynamoDBのサービスから「テーブル作成」をクリックしてください。
次のとおり入力します。

項目	値
名前	tweets
プライマリキー	id（数値） ※id_strなら文字列

入力できたら「作成」をクリックして完了です。

IAMロールの割り当て

EC2からDynamoDBに接続できるように、EC2にIAMロールを割り当てます。
手順は次のとおりです。

1. EC2管理コンソールで今回使うEC2を選択し、「アクション > インスタンスの設定 > IAM ロールの割り当て/置換」をクリックする。
2. 「新しい IAM ロールを作成する」をクリックする。
3. IAM管理コンソールで「ロールの作成」をクリックする。
4. 「AWS サービス > DynamoDB > DynamoDB - Global Tables」を選択し、「次のステップ:アクセス権限」をクリックする。
5. 「AmazonDynamoDBFullAccess」を選択し、「次のステップ:確認」をクリックする。
6. 「ロール名」と「ロールの説明」を入力し、「ロールの作成」をクリックする。
7. EC2管理コンソールに戻って、作成したIAMロールを選択し「適用」をクリックする。

これで対象のEC2はDynamoDBに接続できるようになりました。

Jupyter Notebookでの開発

必要なパッケージのインストール

今回必要なのはrequests_oauthlibだけです。TwitterのOAuth認証に使います。
pip3でインストールしますが、これもJupyter Notebook上でやってしまいましょう。

!sudo pip3 install requests_oauthlib

Twitterへの接続

インストールしたrequests_oauthlibを使ってTwitterに接続します。
OAuth認証キーはここでようやく使います。

from requests_oauthlib import OAuth1Session

CONSUMER_KEY = "取得したConsumer key"
CONSUMER_SECRET = "取得したConsumer secret"
ACCESS_TOKEN = "取得したAccess token"
ACCESS_TOKEN_SECRET = "取得したAccess token secret"
twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

Twitter APIでデータを取得する処理を関数にしておきます。

def get(url, params=None):
    response = twitter.get(url, params=params)
    text = response.text.replace(':""', ':null') # DynamoDBに入れるとき空文字が含まれているとエラーになるため、nullに置換する
    results = json.loads(text, parse_float=decimal.Decimal) # DynamoDBはFloatをサポートしないため、Decimalに変換する
    return results

DynamoDBへの接続

PythonからDynamoDBを使うにはboto3を使います。AWSのDeep Learning AMIなら標準でインストールされていますので、すぐに使えます。
リージョン名とテーブル名は適当に書き換えてください。

import boto3

dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1') # 東京リージョン
tweets_table = dynamodb.Table('tweets') # tweetsテーブル

Twitterの検索結果をDynamoDBに入れる

メインの処理です。今回は次の仕様でデータを取得します。

1. Yokohama（WOEID=1118550）の50件トレンドを取得する。（参考：Twitterのトレンドを取得する際に指定できる日本の都市）
2. トレンド毎にツイートを100件検索する。

TwitterのStandard search APIは15分間に180回しか使えないので、この処理を15分間に3回実行すると制限がかかります。ご注意ください。

import json
import decimal
from multiprocessing import Process

trends = get("https://api.twitter.com/1.1/trends/place.json", params={"id": "1118550"}) # Yokohama

for trend in trends[0]['trends']:
    tweets = get("https://api.twitter.com/1.1/search/tweets.json", params={"q": trend['query'], "count": "100"})
    with tweets_table.batch_writer() as batch:
        for tweet in tweets['statuses']:
            batch.put_item(Item=tweet)

実行してDynamoDBにデータが追加されていれば成功です。

スループットキャパシティーの設定

作ったばかりのDynamoDBでは、今回の処理はとても5分では終わらないと思います。そこでDynamoDBの書き込みのスループットキャパシティーを変更します。
書き込みのスループットキャパシティーは、概ね1秒あたりの書き込み容量[kbytes]で計算できます。（参考：読み取りと書き込みのスループットキャパシティー）
処理が5分で終わるようにする（実際にスループットが出るかはさておき）と、次の式で計算できます。

50[トレンド数]\times 100[検索件数/トレンド]\times 3[kbytes/ツイート（想定）]\div 300[sec]=50[ユニット]

とはいえ、流石に5分フルで処理されると次の処理を圧迫してしまう恐れがあるので、多めに見積もって今回は100[ユニット]とします。
ただし、100[ユニット]はDynamoDBの無料枠を超えていますので、課金が発生します。ご注意ください。
作ったDynamoDBのテーブルを選択して「容量」タブを開き、「書き込み容量ユニット」に100を入力し、問題なければ「保存」をクリックしてください。

処理時間の計測

書き込みのスループットキャパシティーを変更したので、処理時間を計測してみましょう。
Jupyter Notebookでは%%timeというブロック内の処理時間を表示するマジックコマンドが使えます。

%%time
trends = get("https://api.twitter.com/1.1/trends/place.json", params={"id": "1118550"}) # Yokohama

counts = 0
for trend in trends[0]['trends']:
    tweets = get("https://api.twitter.com/1.1/search/tweets.json", params={"q": trend['query'], "count": "100"})
    counts += len(tweets['statuses'])
    with tweets_table.batch_writer() as batch:
        for tweet in tweets['statuses']:
            batch.put_item(Item=tweet)
print("Counts:", counts)

Counts: 4100
CPU times: user 12.9 s, sys: 0 ns, total: 12.9 s
Wall time: 39.7 s

Countsは5000件になる想定でしたが、ちょっと少ないですね。トレンドによってツイート数が100件に満たないのかもしれません。
処理時間は想定（2分半）以上の結果です。

crontabで5分毎に実行

ここからはJupyter Notebookを離れてコマンドラインで操作します。

IPython NotebookからPythonへ変換

コマンドラインで実行するためにIPython Notebookファイル（.ipynb）をPythonファイル（.py）に変換します。
!sudoや%%timeが含まれていると実行時にエラーになるため、コメントアウトしておきましょう。（コメントアウトは変換後でもOKです）
次のコマンドで変換します。

jupyter nbconvert --to=python twitter_crawler.ipynb

コマンドラインでPythonを実行（トラブルシューティングあり）

python3でtwitter_crawler.pyを実行すると次のエラーが起きます。

$ python3 /home/ubuntu/workspace/examples/twitter_crawler.py 
Traceback (most recent call last):
  File "/home/ubuntu/workspace/examples/twitter_crawler.py", line 21, in <module>
    from requests_oauthlib import OAuth1Session
ModuleNotFoundError: No module named 'requests_oauthlib'

requests_oauthlibはJupyter Notebookからpip3でインストールしたはずですが、なぜでしょう。
上記のpython3が何者か調べてみます。

$ which python3
/home/ubuntu/anaconda3/bin//python3

ユーザディレクトリのpython3を見ています。AWSのDeep Learning AMIはLinux標準のPythonではなく、新しいバージョンのPythonを使えるように配慮しているようです。
このパスはどこで設定されているかというと、.profileです。

$ cat ~/.profile

...
# set PATH so it includes user's private bin directories
PATH="$HOME/bin:$HOME/.local/bin:$PATH"
export PATH=$HOME/anaconda3/bin/:$PATH

pip3はどうでしょう。

$ which pip3
/usr/local/bin/pip3

こちらはLinux標準のPythonを見ているようです。つまりpip3でパッケージをインストールしてもpython3が見ているdist-packagesにはインストールされません。
解決方法はいくつかありますが、今回はAWS側のPythonにrequests_oauthlibをインストールすることにします。それにはpipを使えば良いみたいです。

$ which pip
/home/ubuntu/anaconda3/bin//pip
$ pip install requests_oauthlib
Collecting requests_oauthlib
  Downloading requests_oauthlib-0.8.0-py2.py3-none-any.whl
Requirement already satisfied: requests>=2.0.0 in ./anaconda3/lib/python3.6/site-packages (from requests_oauthlib)
Collecting oauthlib>=0.6.2 (from requests_oauthlib)
  Downloading oauthlib-2.0.6.tar.gz (127kB)
    100% |████████████████████████████████| 133kB 5.3MB/s 
Requirement already satisfied: chardet<3.1.0,>=3.0.2 in ./anaconda3/lib/python3.6/site-packages (from requests>=2.0.0->requests_oauthlib)
Requirement already satisfied: idna<2.7,>=2.5 in ./anaconda3/lib/python3.6/site-packages (from requests>=2.0.0->requests_oauthlib)
Requirement already satisfied: urllib3<1.23,>=1.21.1 in ./anaconda3/lib/python3.6/site-packages (from requests>=2.0.0->requests_oauthlib)
Requirement already satisfied: certifi>=2017.4.17 in ./anaconda3/lib/python3.6/site-packages (from requests>=2.0.0->requests_oauthlib)
Building wheels for collected packages: oauthlib
  Running setup.py bdist_wheel for oauthlib ... done
  Stored in directory: /home/ubuntu/.cache/pip/wheels/e5/46/f7/bb2fde81726295a13a71e3c6396d362ab408921c6562d6efc0
Successfully built oauthlib
Installing collected packages: oauthlib, requests-oauthlib
Successfully installed oauthlib-2.0.6 requests-oauthlib-0.8.0

これで冒頭のコマンドが動くようになりました。

$ python3 /home/ubuntu/workspace/examples/twitter_crawler.py 

未解決事項

とりあえず動くようになったので調査は打ち切ってしまいましたが、なぜJupyter Notebookがpip3で動いたのかは謎です。
結果論ですが、Jupyter NotebookはLinux標準のPythonを見ていたようです。
一方で、jupyterはAWSがインストールしたものを使っています。

$ which jupyter
/home/ubuntu/anaconda3/bin//jupyter

crontabの設定

次のコマンドでcrontabを設定します。

$ crontab -e

初めて起動する際はエディタを選べますので、お好きなものを選択してください。
末尾に次の1行を追記してください。

*/5 * * * * python3 /home/ubuntu/workspace/bin/twitter_crawler.py >> /home/ubuntu/workspace/log/twitter_crawler.log 2>&1

保存してエディタを閉じると、次のメッセージが表示されます。

crontab: installing new crontab

これで、5分毎にPythonでTwitterの検索結果をDynamoDBに入れ続けるようになります。
お疲れさまでした。