AWS LambdaでAthenaを呼び出す

[温度センサから値を取得して値を確認する画面をサーバーレスで構築する](https://qiita.com/naoto_koyama/items/2b42e26327210911aaeb)でlambdaでAthenaを叩いてデータを取得するということをやって実際にやってみたので、それのメモ。

##環境
lambda Node.js 6.10

##アーキテクチャ
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/df2773a1-3bc3-dc7d-f5ad-fb437c69d708.png)

##INPUT(S3)
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/1284984a-4557-2436-0199-68306967c97a.png)
上記にtemprerature2018/MM/DD/HH/センサーデータが保存されてます。
センサーデータの例はこんな感じで、JSON形式で保存しています。
humidが湿度、tempが温度、batteryがセンサーのバッテリーです。

```json:センサーデータ
{"timestamp":"2018-02-07T15:19:28.000Z","deviceId":"EBD012965BEB","humid":44,"temp":25.83,"battery":3.18}
{"timestamp":"2018-02-07T15:19:33.000Z","deviceId":"EBD012965BEB","humid":44,"temp":25.81,"battery":3.18}
```

## Athenaの設定
Athenaを開いて左側のCrate tableからManuallyを選択
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/135eb81e-543a-7109-2dfa-f7568493e0cd.png)
Databaseを作成します。入力値を適当に入れてください。
Location of Input Data SetはS3のパスを指定します。
今回temperature2018以下のファイルをIPNUTとするので、そこまでで止めておきます。
(temperature2018/02/08/とかにはしない)
注意点として最後にスラッシュ「/」がないと怒られます。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/d94564bc-4839-f1d7-90c3-c907d998f0c2.png)
S3の入力フォーマットを指定します。今回は上にも書いたようにJSONですね。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/67a78721-3d1a-83db-a398-86e247f161b8.png)
テーブルの列を指定します。
ちょっと検証していないので正しいかは不明ですが、おそらくS3のインプット列すべて指定しないと正しく動いてくれません。batteryを指定するのを忘れていたら動かなかったので。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/d615598a-8e5c-cb81-3d6b-cc2fc56cec3a.png)
Nextをクリックし、Create tableしたらOKです。

正しくSELECT出来ました！
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/a3c37c0d-1456-db69-665c-466e523792b0.png)

Athenaは実行結果を自動でS3にバケットを作成しますので、それのバケット名をメモしておきます（Lambdaの設定で使います）。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/d31ed08b-a4df-ecc4-ce15-791b249cc8f4.png)

## Lambdaの設定

```javascript:
var AWS = require('aws-sdk');
var athena = new AWS.Athena({apiVersion: '2017-05-18'});

function startQuery(){
    return new Promise(function(resolve, reject){
        var params = {
            QueryString: 'SELECT * FROM temperature.temp limit 10;', /* required */
            ResultConfiguration: { /* required */
                OutputLocation: 's3://aws-athena-query-results-XXXXXXX-us-east-1/' // ここにAthenaのアウトプットのパスを指定する
            }
        };
        athena.startQueryExecution(params, function(err, data) {
            console.log('data : ' , data);
            resolve(data.QueryExecutionId);
        });
       
    });
}

function waitQueryEnd(queryExecutionId){
    return new Promise(function(resolve, reject){
        var time = 0;
        var params = {QueryExecutionId: queryExecutionId /* required */};
        (function loop() {
            if (time != 20) {
                athena.getQueryExecution(params, function(err, data) {
                    console.log('Status : ' , data.QueryExecution.Status.State);
                    if (data.QueryExecution.Status.State == 'SUCCEEDED'){
                        resolve(queryExecutionId);
                    } else {
                        sleep(2000);
                        time+=2;
                        loop();
                    }
                });
            } else {
                reject('timeout error!!!');
                return false;
            }
        }());
        
        
    });
}

function sleep(waitMsec) {
 
    var startMsec = new Date();

    // 指定ミリ秒間、空ループ。CPUは常にビジー。
    while (new Date() - startMsec < waitMsec);

}

function getResult(queryExecutionId){
    return new Promise(function(resolve, reject){
        var params = {
          QueryExecutionId: queryExecutionId, /* required */
        };
        athena.getQueryResults(params, function(err, result) {
            var response = [];
            for(let i = 1; i < result.ResultSet.Rows.length; i++) {
              var row = result.ResultSet.Rows[i].Data;
              var data = 
                  { timestamp : row[0].VarCharValue,
                    deviceid : row[1].VarCharValue,
                    humid : row[2].VarCharValue,
                    temp : row[3].VarCharValue,
                    battery : row[4].VarCharValue
                  };
                response.push(data);
                
            }
            resolve(response);
        });
        
    });
}

exports.handler = (event, context, callback) => {
    
    console.log('event : ' , event);
    var promise = Promise.resolve();
    promise
        .then(startQuery)
        .then(waitQueryEnd)
        .then(getResult)
        .then(function(value){
            callback(null,{statusCode: 200,
                      headers: {
                        "Access-Control-Allow-Origin" : "*" // Required for CORS support to work
                      },
                      body : value});
        })
        .catch(function(error){
            callback(null,error);
        });
    
};

```

SDKを使って取得しています。またPromiseのメソッドチェーンを使っています。
Promiseの詳細は[Promiseのメソッドチェーンの使い方まとめ](https://qiita.com/naoto_koyama/items/7234d133f0d5f87532db)参照。

###startQuery
Athenaを呼び出して、SQLを実行しています。
startQueryExecutionは実行するだけなので、実行結果が返ってくるわけではありませんので、注意。
アウトプットはAthenaの設定でメモしておいたところを設定します。
クエリのIdを次の関数へ渡しておきます。

###waitQueryEnd
クエリが終了するのを待ちます。
20秒以上クエリに時間がかかったらエラーとしています。

###getResult
クエリの結果を取得しています。
クエリの結果が以下のような形になるため、データを整形して返しています。

```json
[{"Data":[{"VarCharValue":"timestamp"},{"VarCharValue":"deviceid"},{"VarCharValue":"humid"},{"VarCharValue":"temp"},{"VarCharValue":"battery"}]}
,{"Data":[{"VarCharValue":"2018-02-07T03:46:32.000Z"},{"VarCharValue":"EBD012965BEB"},{"VarCharValue":"41.0"},{"VarCharValue":"25.88"},{"VarCharValue":"3.18"}]}
,{"Data":[{"VarCharValue":"2018-02-07T03:46:42.000Z"},{"VarCharValue":"EBD012965BEB"},{"VarCharValue":"41.0"},{"VarCharValue":"25.93"},{"VarCharValue":"3.18"}]}
]
```

###その他
API GatewayからLambdaを呼び出すときにクロスドメインでエラーになってしまうので、
callback関数にレスポンスヘッダーを追記しています。

またデフォルトではLambdaのタイムアウト設定が3秒になっているため、20秒にしておきます。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/202761/7abca1e6-4a7d-1bf5-8ce6-83d512886347.png)

##実行結果

```json
{"statusCode":200
,"headers":{"Access-Control-Allow-Origin":"*"}
,"body":[{"timestamp":"2018-02-06T18:56:37.000Z","deviceid":"EBD012965BEB","humid":"41.0","temp":"25.79","battery":"3.18"}
		,{"timestamp":"2018-02-06T18:56:42.000Z","deviceid":"EBD012965BEB","humid":"41.0","temp":"25.79","battery":"3.18"}
		,{"timestamp":"2018-02-06T18:56:57.000Z","deviceid":"EBD012965BEB","humid":"41.0","temp":"25.78","battery":"3.18"}
		]}
```

##まとめ
関数の詳細な説明については以下を参考にしてください。
https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/Athena.html