Lambdaで取得したTwitterの情報をElasticsearchとKibanaで可視化したらすごいことになった話・前編

1.はじめに

こんにちは！毎回記事が長いと言われる山内です。今回も長くなってしまったので、前編と後編に分けてお届けいたします！お付き合いください。

タイトルをご覧いただければお分かりになるかと思いますが、最近、Twitterの情報を見える化する機会があったので、今回はそのお話を書かせていただきます。

SNSが浸透してきたこの世の中、「エゴサーチ」ってしたことある人が多いんじゃないでしょうか。エゴサーチをしたことがない人も、SNSを使っていると、トレンドワードみたいなものが表示された経験がある人がほとんどかと思います。筆者も、最新のニュースなどはTwitterのトレンドで知ることが結構あります。

そこで、たくさんあるTwitterの情報を可視化したら楽しそうではないか！ということで、やってみました！(*1)もし、「作る過程は良いから、作ったものが見たい！」という方がいたら、後編の「10.楽しむ！」までジャンプしてください。

（余談ですが、TwitterはSNSじゃないそうです。Twitter社自身も明確に否定しているみたいなので、興味がある方は「Twitter SNS ではない」とかでググってみてくださいね。）

*1 AWSマネジメントコンソールの画面は2019/05/21時点のものです。

2.今回作ったもの

今回作ったものは、Twitterから特定のキーワードで検索したものを、ElasticsearchとKibanaで可視化する、といったものです。この際、AI関連のサービスであるAmazon ComprehendやAmazon Translateを活用し、ツイートの感情分析（ポジネガ分析）も行えるようにしています。本当はComprehendだけでやりたかったのですが、Comprehendが日本語に対応しておらず、Translateで英語に翻訳する処理を入れました。（このときは、まさかあんなにすごいことになるとは思いもしなかった…。）

それから、構成図はこんな感じです。

まず、Twitter APIを利用するための準備として、APIへのアクセス情報を取得する必要があります。下記の手順で、アクセス情報を取得してください。

1. TwitterのDeveloper Accountを取得する。（取得の仕方は既にネット上にたくさん記事があるので、そちらをご参照ください。）

2. 下記4つの項目をメモする。

   • API key
   • API secret key
   • Access token
   • Access token secret

   

次に、今回のテーマである「可視化」を行うためのデータを蓄積するElasticsearchを構築していきます。

1. AWSマネジメントコンソールで「Elasticsearch Service」を選択する。
2. 「新しいドメインの作成」をクリックする。
3. 下記情報を入力し、「次へ」をクリックする。
   • デプロイタイプ　：　任意（今回は「開発およびテスト」）
   • バージョン　　　：　任意（今回は「6.5」）

   

4. 下記情報を入力し、「次へ」をクリックする。
   • ドメイン名　　　　　：　任意（今回は「sns-analysis-demo」）
   • インスタンスタイプ　：　任意（今回は「t2.medium.elasticsearch」）

   

5. 下記情報を入力し、「次へ」をクリックする。
   • ネットワーク構成　：　任意（今回は「パブリックアクセス」）
   • アクセスポリシー　：　任意（今回は「テンプレートを選択」→「ドメインへのオープンアクセスを許可」）

   ※今回の設定はかなりのセキュリティリスクを伴うため、本来は設定すべきではありません。今回はデモ用の簡易的な環境構築のため、上記設定を行っています。実際はアクセス制限など、しっかり設定してください。

6. 設定した項目を再度確認し、「確認」をクリックする。
7. ドメインのステータスが「アクティブ」になればOK。

1. AWSマネジメントコンソールで「AWS Lambda」を選択する。
2. 「関数の作成」をクリックする。
3. 下記情報を入力し、「関数の作成」をクリックする。
   • 関数名　　　：　任意（今回は「sns-analysis-demo」）
   • ランタイム　：　Node.js 10.x

   

4. 下記コードを転記し、右上の「保存」をクリックする。※requireしているモジュールを合わせてZIPファイルをアップロードする。
   sns-analysis-demo

   JavaScript

   /** モジュールのimport */ const twitter = require('twitter'); const elasticSearch = require('elasticsearch'); const AWS = require('aws-sdk'); /** 各クライアントの生成 */ // elasticsearch用のクライアント const elasticsearchClient = new elasticSearch.Client({ service : 'es', region : process.env.ES_REGION, host : process.env.ES_HOST }); // Twitter用のクライアント const twitterClient = new twitter({ consumer_key : process.env.TW_CONSUMER_KEY, // API Key consumer_secret : process.env.TW_CONSUMER_SECRET, // API Secret Key access_token_key : process.env.TW_ACCESS_TOKEN_KEY, // Access Token access_token_secret : process.env.TW_ACCESS_TOKEN_SECRET, // Access Token Secret }); // 翻訳用 const translate = new AWS.Translate(); const translateParams = { SourceLanguageCode: 'ja', TargetLanguageCode: 'en', Text: '' }; // 感情分析用 const comprehend = new AWS.Comprehend({region: 'us-east-1'}); /** * Elasticsearchにデータを登録します。 */ var createIndexAndStreamDataOnES = function(indexName, id, body) { // Elasticsearchに登録 elasticsearchClient.create({ index : indexName, type : 'test', id : id, body : body }, function(err, data) { if (err) { console.log("Error on creating data", err); } else { console.log("data reply received", data); } }); }; /** * 与えられた文字列の感情分析をします。 */ const detectSentiment = function (text) { return new Promise((resolve, reject) => { let params = { LanguageCode: "en", Text: text }; comprehend.detectSentiment(params, function (err, data) { if (err) { console.log(err, err.stack); reject(err); } else { resolve(data.Sentiment); } }); }); }; /** * 引数に与えられた文字列を含むTweetを検索し、 * 検索結果をElasticsearchに登録します。 */ function searchTweet(queryArg, nextResultsMaxIdArg = null) { // Tweetを検索 twitterClient.get('search/tweets', { q: queryArg, count: 100, max_id: nextResultsMaxIdArg }, (error, searchData, response) => { // エラーが発生した場合 if (error) { console.log(error); return 1; } // Tweetの検索結果が存在していた場合 if (searchData !== undefined) { var item = null; for (item in searchData.statuses) { // tweet情報 let tweet = searchData.statuses[item]; // タイムスタンプ情報を保持 tweet["@timestamp"] = new Date(tweet["created_at"]); // 検索条件を保持 tweet["searchParam"] = queryArg; // 英語に翻訳 translateParams["Text"] = tweet.text; translate.translateText(translateParams, function(err, translateResult) { if (err) { console.log(err, err.stack); // an error occurred } else { // 英訳したテキストの感情を分析 detectSentiment(translateResult.TranslatedText).then(function(sentiment) { // 分析結果を保持して、elaseticsearchに登録 tweet["sentiment"] = sentiment; createIndexAndStreamDataOnES("twitter", tweet["id_str"], tweet); }); } }); } if (searchData.search_metadata == undefined) { return 0; } else if (searchData.search_metadata.next_results) { let maxId = searchData.search_metadata.next_results.match(/\?max_id=(\d*)/); if (maxId[1] == null) { return 0; } searchTweet(queryArg, maxId[1]); } else { return 0; } } }); } /** * Lambdaのハンドラー関数です。 */ exports.handler = function (event) { // ツイートを検索 searchTweet(event.keyword); // レスポンスを返す const response = { statusCode: 200, body: JSON.stringify('Success!'), }; return response; };

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143

   /** モジュールのimport */ const twitter       = require('twitter'); const elasticSearch = require('elasticsearch'); const AWS           = require('aws-sdk'); /** 各クライアントの生成 */ // elasticsearch用のクライアント const elasticsearchClient = new elasticSearch.Client({     service : 'es',     region  : process.env.ES_REGION,     host    : process.env.ES_HOST }); // Twitter用のクライアント const twitterClient = new twitter({   consumer_key        : process.env.TW_CONSUMER_KEY,        // API Key   consumer_secret     : process.env.TW_CONSUMER_SECRET,     // API Secret Key   access_token_key    : process.env.TW_ACCESS_TOKEN_KEY,    // Access Token   access_token_secret : process.env.TW_ACCESS_TOKEN_SECRET, // Access Token Secret }); // 翻訳用 const translate       = new AWS.Translate(); const translateParams = {   SourceLanguageCode: 'ja',   TargetLanguageCode: 'en',   Text: '' }; // 感情分析用 const comprehend = new AWS.Comprehend({region: 'us-east-1'}); /** * Elasticsearchにデータを登録します。 */ var createIndexAndStreamDataOnES = function(indexName, id, body) {   // Elasticsearchに登録   elasticsearchClient.create({     index : indexName,     type  : 'test',     id    : id,     body  : body   }, function(err, data) {       if (err) {           console.log("Error on creating data", err);       } else {           console.log("data reply received", data);       }   }); }; /** * 与えられた文字列の感情分析をします。 */ const detectSentiment = function (text) {   return new Promise((resolve, reject) => {     let params = {       LanguageCode: "en",       Text: text     };     comprehend.detectSentiment(params, function (err, data) {       if (err) {         console.log(err, err.stack);         reject(err);       } else {         resolve(data.Sentiment);       }     });   }); }; /** * 引数に与えられた文字列を含むTweetを検索し、 * 検索結果をElasticsearchに登録します。 */ function searchTweet(queryArg, nextResultsMaxIdArg = null) {   // Tweetを検索   twitterClient.get('search/tweets', { q: queryArg, count: 100, max_id: nextResultsMaxIdArg }, (error, searchData, response) => {     // エラーが発生した場合     if (error) {       console.log(error);       return 1;     }     // Tweetの検索結果が存在していた場合     if (searchData !== undefined) {       var item = null;       for (item in searchData.statuses) {         // tweet情報         let tweet = searchData.statuses[item];         // タイムスタンプ情報を保持         tweet["@timestamp"] = new Date(tweet["created_at"]);         // 検索条件を保持         tweet["searchParam"] = queryArg;         // 英語に翻訳         translateParams["Text"] = tweet.text;         translate.translateText(translateParams, function(err, translateResult) {           if (err) {             console.log(err, err.stack); // an error occurred           } else {             // 英訳したテキストの感情を分析             detectSentiment(translateResult.TranslatedText).then(function(sentiment) {               // 分析結果を保持して、elaseticsearchに登録               tweet["sentiment"] = sentiment;               createIndexAndStreamDataOnES("twitter", tweet["id_str"], tweet);             });           }         });       }       if (searchData.search_metadata == undefined) {         return 0;       } else if (searchData.search_metadata.next_results) {         let maxId = searchData.search_metadata.next_results.match(/\?max_id=(\d*)/);         if (maxId[1] == null) {           return 0;         }         searchTweet(queryArg, maxId[1]);       } else {         return 0;       }     }   }); } /** * Lambdaのハンドラー関数です。 */ exports.handler = function (event) {   // ツイートを検索   searchTweet(event.keyword);   // レスポンスを返す   const response = {       statusCode: 200,       body: JSON.stringify('Success!'),   };   return response; };

5. 「基本設定」内の下記項目を設定し、「保存」をクリックする。
   • 説明　　　　　：　任意
   • タイムアウト　：　15分

   

6. 「環境変数」に下記情報を登録する。
   • ホスト
     • キー　：　ES_HOST
     • 値　　：　（Elasticsearchの「エンドポイント」の「https://」を除いたもの
   • Elasticsearchを作成したリージョン
     • キー　：　ES_REGION
     • 値　　：　（今回は「ap-northeast-1」）
   • Twitter APIへのアクセス情報（API Key）
     • キー　：　TW_CONSUMER_KEY
     • 値　　：　TwitterのDeveloper Account作成時にメモしたもの（API Key）
   • Twitter APIへのアクセス情報（API Secret Key）
     • キー　：　TW_CONSUMER_SECRET
     • 値　　：　TwitterのDeveloper Account作成時にメモしたもの（API Secret Key）
   • Twitter APIへのアクセス情報（Access Token）
     • キー　：　TW_ACCESS_TOKEN_KEY
     • 値　　：　TwitterのDeveloper Account作成時にメモしたもの（Access Token）
   • Twitter APIへのアクセス情報（Access Token Secret）
     • キー　：　TW_ACCESS_TOKEN_SECRET
     • 値　　：　TwitterのDeveloper Account作成時にメモしたもの（Access Token Secret）

   

7. 「実行ロール」内の「~~~~ロールを表示」リンクをクリックする。
8. 「ポリシーをアタッチします」をクリックする。
9. 下記ポリシーをチェックし、「ポリシーのアタッチ」をクリックする。
   • • ConprehendReadOnly
     • TransolateReadOnly

   

ここまででElasticsearchとLambdaの作成は終わりです。次回、後編ではKibanaを設定するところからご説明いたします。お楽しみに！

※TWITTER、TWEET（ツイート）、RETWEET（リツイート）、Twitter のロゴはTwitter, Inc.またはその関連会社の登録商標です。
※KibanaはElasticsearch BVの登録商標です。

Lambdaで取得したTwitterの情報をElasticsearchとKibanaで可視化したらすごいこと...

https://www.tdi.co.jp/miso/lambda-twitter-elasticsearch-kibana-2

SNSを使っていると、トレンドワードが表示された経験がある人がほとんどかと思います。筆者も、最新のニュースなどはTwitterのトレンドで知ることが結構あります。そこで、たくさんあるTwitterの情報を可視化したら楽しそうではないか！ということで、やってみました！

MISO｜未来を創造するITのミソ

 

3 Pockets