RDSに蓄積しているアクセスログを今すぐDataLakeに移行する

## DataLakeとは

https://aws.amazon.com/jp/big-data/datalakes-and-analytics/what-is-a-data-lake/
AWS公式の記述を引用すると

> データレイクは、規模にかかわらず、すべての構造化データと非構造化データを保存できる一元化されたリポジトリです。データをそのままの形で保存できるため、データを構造化しておく必要がありません。また、ダッシュボードや可視化、ビッグデータ処理、リアルタイム分析、機械学習など、さまざまなタイプの分析を実行し、的確な意思決定に役立てることができます。

　当記事ではDataLakeのデータストアをS3とし、既存のRDSに蓄積されている既存ログ(レガシー)をGlueを使ってS3に移行する手順を紹介します。
　なお、S3に移行したログの可視化や活用方法は当記事の対象範囲外です。
　（ただしAthenaを使ってログを抽出できることは確認します）

　また、今後蓄積されるログはRDSではなくS3に蓄積し続けることになりますが、当記事では既存ログの移行方法を紹介するまでに留めます。
　（やるとすればKinesisFirehoseを使うことになるでしょう）

## 構成図

![DataLake.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/1fc550a4-220e-4ecc-2481-995f64c87873.png)

RDSはVPC内のプライベートサブネットに配置されています。

- クローリング
  GlueからVPC内のRDSへ接続しクローリングを行ないデータカタログを作ります。

- ジョブ
  S3へ物理的にログを移行します

- クエリ発行
  S3へ移行されたログを抽出してみます

## GlueからRDSへ接続する

#### Glue設定

まずはGlueの`接続`より`接続の追加`を押下します。
<img width="1378" alt="スクリーンショット 2019-09-07 19.51.06.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/f65ad4ca-94e4-c1b3-af7d-b7218117a143.png">

レガシーログが格納されているRDSを選択し、画面通りに設定を進めて完了します。
<img width="1381" alt="スクリーンショット 2019-09-07 19.52.20.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/4d8b963b-803c-f5d1-f106-c39bac2e1a59.png">


#### セキュリティグループ作成

RDSをVPCに配置している場合はセキュリティグループが必要です。
以下を参考に、Glueの自己参照型セキュリティグループを作成します。
https://docs.aws.amazon.com/ja_jp/glue/latest/dg/setup-vpc-for-glue-access.html

全てのトラフィックを、 **自身のセキュリティグループからのみ呼ばれるように** 設定します。
<img width="1162" alt="スクリーンショット 2019-09-07 20.01.36.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/4586e6f5-670c-69a8-2a53-c64bcba5fec9.png">

このセキュリティグループを、RDSのセキュリティグループに全てのTCPを開放して設定します。（以下はRDSにアタッチされているセキュリティグループの例）
<img width="1166" alt="スクリーンショット 2019-09-07 20.02.10.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/4e1b11ed-879c-1b60-c4c7-803eca4e066b.png">

#### 接続テスト

ここまで進めればGlueの`接続`より`接続テスト`をしてみましょう。
ここで失敗すれば設定等に不備がある可能性があります。見直しましょう。

## クローリング①

#### データベース作成

Glueからデータベースの作成を行ないます。
![スクリーンショット 2019-09-08 23.36.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/bb4d5f8a-1816-bff0-fbdc-a3e85f4b1d58.png)

#### クローリング実行

続いてクローラを作成し、実行します。
![スクリーンショット 2019-09-08 23.38.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/c67e869a-3f0b-a13e-8222-4d7e03eebe71.png)

データストアは`JDBC`を選択し、先ほど作成した接続を選択します。
実行は`オンデマンドで実行`とし、最後に手動実行します。
![スクリーンショット 2019-09-08 23.39.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/5f8df8e7-296f-f750-2330-eb8803727b7b.png)

実行後、左メニューの`テーブル`にテーブルが追加されていることを確認しましょう。

## S3バケット作成

Glueの`テーブル`にクローリングの結果が格納されていますが、物理的なログの格納場所はS3となります。
この次の手順でジョブを作成しログを流し込みますが、そのためのS3バケットを事前に作成します。
適当なバケット名を指定して作成しておきましょう。

## Glueジョブ作成

続いてS3へ物理的にログを移行するジョブを作成します。
<img width="1377" alt="スクリーンショット 2019-09-07 20.19.22.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/e7b6d511-5d51-9a11-ab2f-783fb68453be.png">

ジョブにはスクリプトを作成する必要があります。
今回は`AWS Glue が生成する提案されたスクリプト`を選択して進めます。
<img width="1383" alt="スクリーンショット 2019-09-07 20.15.57.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/3ea836c3-e400-65bb-a63f-1d38efc94e25.png">

なお、IAMロールを作成する必要があるので存在しない場合は新規で作成します。
アタッチするポリシーは`AWSGlueServiceRole`と`AmazonS3FullAccess`にします。

<img width="995" alt="スクリーンショット 2019-09-07 20.23.40.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/11062c97-62e6-46cd-b122-c51dcc2c7ea5.png">

その他オプションについては自由ですが、モニタリングオプションは有効化しておくのがおすすめです。
エラーになった場合に原因が特定されやすくなります。
<img width="1380" alt="スクリーンショット 2019-09-07 20.16.27.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/7c9f6311-8e09-87b0-16ed-455df5308852.png">

そして設定を進めていくとスクリプトの編集画面に進みます。

```python

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

〜（省略）〜

## @type: DataSink
## @args: [database = "xxxxx", table_name = "xxx_logs", transformation_ctx = "datasink5"]
## @return: datasink5
## @inputs: [frame = resolvechoice4]

# datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = resolvechoice4, database = "xxxxx", table_name = "xxx_logs", transformation_ctx = "datasink5")
datasink5 = glueContext.write_dynamic_frame.from_options(frame = resolvechoice4, connection_type = "s3", connection_options = {"path": "s3://xxx/yyy/zzz"}, format = "parquet", transformation_ctx = "datasink5")
job.commit()
```

最終行付近にある`glueContext.write_dynamic_frame.from_catalog`の部分を編集します。
今回はS3に配置するため、以下のようにします。

```python
glueContext.write_dynamic_frame.from_options(frame = resolvechoice4, connection_type = "s3", connection_options = {"path": "s3://xxx/yyy/zzz"}, format = "parquet", transformation_ctx = "datasink5")
```

`s3://xxx`がバケット、`yyy/zzz`はキー(プレフィックス)になりますが、S3ではキーで区切ることでデータストアのパーティションとして定義することができます。パーティションについての解説は当記事の対象範囲外とします。

## クローリング②

ジョブの実行が完了すると、S3に`.parquet`形式のログデータが格納されていると思います。
このデータをデータカタログとして登録するためにクローラの作成を行ないます。

前回の手順ではデータストアをRDSとしましたが、今回はS3です。
![スクリーンショット 2019-09-09 9.26.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/18ad9f2c-ced4-a8b4-adf9-8e196c99a9db.png)

クローラを実行し、左メニューの`テーブル`にS3をデータストアとしたテーブルが追加されていることを確認しましょう。

なお、クローラを実行しないとデータカタログが更新されませんので
商用で活用する場合は定期的に実行するようにスケジュールを設定することをおすすめします。

## Athenaによるクエリ実行

Athenaによるログの抽出は簡単です。SQLとほぼ同じ形式で書けます。
クローリングしたデータが抽出できることを確認しましょう。

※下記はAWSがあらかじめ用意したサンプルで例を提示しています。
![スクリーンショット 2019-09-09 9.20.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/466431/a41a88cc-7a83-b8db-fe0a-81d954114a68.png)