巨大なテーブルをBigqueryに転送する方法

# Motive
　[twitter分析ページ](https://featurepoints.jp/)を作成して約2年が経ちました。従来はTwitterAPIで取得したデータをMySQLにストックしていましたが、 あるテーブルは約2億レコードと膨大になってしまったため`order by` でなくても`select count(*) from TWEET` などといったシンプルなSQLでも約数十分かかります。システム構造もEC2にMySQLを導入したサーバーにデータを追加してたので、仮に `/etc/my.cnf` でメモリサイズを最大化し、EC2自体の `t3.micro` から `t3.large` に変更しても限界があります。 <br>
　それで解決方法としてはGoogle Cloud Service の Bigquery を使うこととしました。利点としては数テラのかなり大きめのテーブルでも約数１０秒で結果が出力されることです。金額はクエリを使うごとに発生はして、数テラのかなり大きめのテーブルを全てのカラムを使ってクエリをかけた場合は結構かかります。しかし、数カラムだけを使ってwhereで使うデータを制限すれば金額がかからないので問題ないと思っています。
ここで問題になるのは数億あるテーブルをどのようにBigqueryへ移行するかです。 普通であれば`mysqldump`でjsonにoutputしてGCPのストレージに転送すればいいのですが、jsonを作成するのにかなり時間がかかることとテーブルの中にツイート文を扱っているので特殊文字が対応しない可能性がありそうです。ここでは、大規模なデータ処理が可能な`pyspark`を使ってファイルを出力し、Bigqueryへ転送する方法を紹介したいと思います。

# Preparetion

### pyspark の導入方法
anaconda上でpysparkをインストールしました。

```
conda install -c conda-forge pyspark
```

### AWS EC2
データ移行前のEC2のタイプは`t3.micro`を使っていましたがこのままではメモリ不足になってしました。結果としては、`t3.xlarge`(メモリ16GB)までアップグレードしないと無理だったです。いくらpyspark内で分散処理をしようとしても巨大なテーブルをメモリ上にのせる必要があったみたいです。

### GCP Bigquery

移行先のテーブルに追加するデータセットを作成します。リージョンは適宜決める。

<img width="300" alt="dataset" src="https://user-images.githubusercontent.com/41592414/158300061-1a7bbaf5-846b-4cbd-9cf3-c393b80310e6.png">


# Method

下記の流れでデータ移行しました。
ひとつのテーブルに対してファイルをひとつ作るのは無理だったので、データ取得した日のタイムスタンプを使って1ヶ月単位に分割してファイルを出力しました。

1. [GCP]bigqueryで空のテーブルを作成する
2. [AWS]pysparkを使ってデータをparquet形式のファイルとしてoutputする
3. [AWS]parquetファイルをS3に転送する
4. [GCP]bigqueryのデータ転送を使って設定を作成する

![システム構成_2022-データ取り込み.drawio.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/288206/8fb10482-b351-ef8e-6da7-e56d307f8039.png)


# Detail

### [GCP]bigqueryで空のテーブルを作成する

1. テーブルを追加したいデータセットから「テーブルを作成」を選ぶ
<img width="200" alt="capture_01" src="https://user-images.githubusercontent.com/41592414/158300740-cf514c66-3cf0-4bcc-aaf9-08e2a7eabaa1.png">

2. カラムとテーブル名を入力してテーブルを作成する
<img width="800" alt="capture_02" src="https://user-images.githubusercontent.com/41592414/158301202-d36f5ba7-287d-4b81-be30-3b8da1683900.png">

- テーブルの作成元:空のテーブル
- テーブルタイプ:ネイティブテーブル

### [AWS]pysparkを使ってデータをparquet形式のファイルとしてoutputする

```Python
from pyspark.sql import SparkSession

import argparse
import os
import time

JDBC_URL = "jdbc:mysql://[***データベース ホスト名***]/[***データベース データベース名***]"
DRIVE_CLASS = "com.mysql.cj.jdbc.Driver"
DB_USER = "[***データベース ユーザ名***]"
DB_PASSWORD = "[***データベース パスワード***]"

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('table_name', help='テーブル名')
    parser.add_argument('begin', help='where from')
    parser.add_argument('end', help='where to')
    parser.add_argument('output', help='出力されるフォルダ名')
    args = parser.parse_args()

    start = time.time()

    print(f"{args.table_name}:{args.begin}:{args.end}")
    
    #pyspark 初期設定
    spark = SparkSession.builder.config("spark.jars", "/usr/share/java/mysql-connector-java.jar")\
            .config("spark.driver.memory", "2g")\  #メモリ設定
            .master("local").appName("PySpark_MySQL_test")\
            .getOrCreate()

    #MySQLに接続し、ターゲットテーブルをもとにデータフレームを作成
    df = spark.read.format("jdbc").option(
        "url", JDBC_URL).option(
        "driver", DRIVE_CLASS).option(
            "user", DB_USER).option(
                "password", DB_PASSWORD).option("dbtable", args.table_name).load()
    
    #範囲指定する
    df_filtered = df.filter(f"'{args.begin}' <= created_at and created_at < '{args.end}'")
    print(df_filtered.count())

    #parquetファイルを作成
    df_filtered.write\
            .parquet(os.path.join(os.path.dirname(os.path.abspath(__file__)), args.output))

    elapsed_time = (time.time() - start) / 60
    print(f"処理時間:{elapsed_time}分")

if __name__ == "__main__":
    main()

```

ポイントとしてはpysparkの設定時にドライバのメモリを設定する部分です。 デフォルトは1GBなので実行環境内でのメモリを増設している場合はそれ相応の数字にする必要があります。この時は前述の通りEC2`t3.xlarge`でメモリが16GBあったので12GBに設定しました。
また、一度にファイル作成すると膨大に時間がかかるかメモリがオーバーフローするので、whereで数ヶ月単位で区切ってparquetファイルを出力しています。

### [AWS]parquetファイルをS3に転送する

aws cli を使ってparquetファイルをS3に転送します。

```
aws s3 cp XXXXXXXXX.parquet s3://[バケット名]/XXXXX/ 
```

awsのコマンドがなければ `pip3 install awscli` でpipインストールします。

### [GCP]bigqueryのデータ転送を使って設定を作成をする

S3のフォルダに格納したparquetファイルをBigqueryに転送します。Bigqueryには「データ転送」の機能があるのでこれを使います。

1. 「データ転送」を選ぶ
<img width="250" alt="capture_03" src="https://user-images.githubusercontent.com/41592414/159144836-90ed8f6f-14fe-45dc-8d7c-bcf0a439f58e.png">

2. 「転送を作成」を選ぶ
<img width="300" alt="capture_04" src="https://user-images.githubusercontent.com/41592414/159145102-e8cf602b-da48-426d-9d2e-278bf25bc873.png">

3. 設定を記述して「保存」を押す

|カテゴリ|項目|選択項目 or 記述例|
|:--:|:--:|:--|
|ソースタイプ|ソース|Amazon S3 (固定)|
|転送構成名|表示名|sample_s3_to_gcp|
|スケジュールオプション|繰り返しの頻度|オンデマンド　(固定)|
|転送先の設定|データセット| <対象のデータセット> |
|データソースの詳細|Destination Table| <対象のテーブル> |
||Amazon S3 URL| <送信元S3のparquetファイルリンク> |
||Access key ID|<IAMで作成したアクセスキーのID>|
||Secret access key|<IAMで作成したシークレットアクセスキー>|
||File Format| PARQUET(固定) |

このときに事前にIAM(Identity and Access Management)からユーザ作成とAmazonS3FullAccessの権限が付与されたIAMロールが必要になります。「Access key ID」「Secret access key」の取得方法については、[AWS アクセスキーを作成する](https://qiita.com/miriwo/items/a513c91eac7186db59fe)を参照し、csvファイルをダウンロードしてください。

<img width="400" alt="capture_05" src="https://user-images.githubusercontent.com/41592414/159145110-b718c6b0-11aa-4dbd-adb5-eb7be475dd96.png">

4. リストから作成した設定を選び、「今すぐ転送を実行」を選んで転送を実行する。

<img width="500" alt="capture_06" src="https://user-images.githubusercontent.com/41592414/159154440-67be5562-2d5b-4272-8a8b-f17ca8e86415.png">

ここまでできれば転送されるまで少し待てば転送先のテーブルにデータ移行ができてると思います。

# PostScript
今回はS3を経由してデータ転送する方法を紹介しましたが、後ほど試した方法でGCPのGoogle Cloud Storage経由でデータ転送すると空のテーブルを手動で作成することなく、自動でテーブルにデータ転送されるみたいです。次回はGoogle Cloud Storage経由でデータ転送する方法を紹介したいと思います。


# Reference

- [Amazon EC2 T3 インスタンス](https://aws.amazon.com/jp/ec2/instance-types/t3/)
- [How to convert an 500GB SQL table into Apache Parquet?](https://stackoverflow.com/questions/41498672/how-to-convert-an-500gb-sql-table-into-apache-parquet)
- [Spark Parquet ファイル](https://mogile.web.fc2.com/spark/sql-data-sources-parquet.html)
- [How to set spark.driver.memory in client mode - pyspark (version 2.3.1)](https://stackoverflow.com/questions/53606756/how-to-set-spark-driver-memory-in-client-mode-pyspark-version-2-3-1/53620273)
- [S3からGCSへのデータ転送（AWSのIAM情報もらうパターン）](https://note.com/ucwork/n/n5fe04382409b#P8K6j)
- [【入門】AWSのIAMと権限の考え方〜前編〜](https://qiita.com/yut0201/items/3051757362e41e971dec)
- [【入門】AWSのIAMと権限の考え方〜後編〜](https://qiita.com/yut0201/items/16bdc51eefcd41c18958)