CSV取り込みの高速化

## はじめに
クライアントからCSVをFTPでもらい、DBに取り込む機能はシステムでは、良くあるユースケースだと思います。

サービス開始当初は、クライアントの規模も小さく、CSV取り込みの処理時間を気にする必要ないのですが、サービスが大きくなると、処理時間が長くなってテコ入れする必要があるのは、よくあるケースだと思います。

ここでは、CSV取り込みの高速化に対するアプローチを紹介します。

## ETL
CSV取り込み処理は、設計のアプローチとしては、ETL（Extract/Transform/Load）になると思います。

```mermaid
flowchart LR
    subgraph 書き出し/Load
       upsert(DB登録)
    end
    subgraph 変換/Transform
       Transform(CSVレコードパーズ) --> upsert
    end
    subgraph 抽出/Extract
       csv(CSV読み込み) --> Transform
    end
```

参考までにETLのライブラリは以下になります。
- Embulk (https://www.embulk.org/)
- Apache Airflow (https://airflow.apache.org/)
- Luigi (https://github.com/spotify/luigi)
- AWS Glue (https://aws.amazon.com/jp/glue/)

ただ、どのライブラリを選んでもインフラをどうするか？と言う問題が発生します。
AWSだと、`Fargate(ECS/EKS)`、`EC2`のどちらを選んでも定義できるvCPU数（限界）は決まってしまいます。
| 種類 | CPU(vCPU) |
|:-:|:-:|
|`ECS(EKS) on Fargate`|16|
|`ECS(EKS) on EC2`| 96 |
|`EC2`|96| 

このため、クラスタ機能をもつETLツールを選ぶ必要です。

## 抽出(Extract)
- このPhaseでは、CSV読み込みの高速化を考えます。
- CSV読み込みは、言語の性能をもろに受けます。ETLライブラリを選定するときの参考になると思います。
- 以下はChatGPT4Proの回答です。

|言語|ライブラリ/実装|処理時間 (秒)|
|:-:|:-:|:-:|
|`Go`|csv.Reader（並列処理あり）|1.5 - 2.0 秒|
|`Rust`|csv クレート（並列処理あり）|1.0 - 1.5 秒|
|`Python`|pandas.read_csv()（Cエンジン）|2.0 - 3.0 秒|
|`Java`| opencsv / univocity-parsers |1.8 - 2.5 秒|
|`Ruby`| CSV.read（純粋な Ruby 実装）|5.0 - 10.0 秒|

- このため、Extract部分だけ別言語にするのも選択肢になると思います。

### 参考
- [PythonとGolangのCSVの処理速度の違いを調べる](https://nantekottai.com/2021/10/23/python-golang-csv-performance/)
- [Rustは本当に動作が高速なのか？ Pythonとの比較で分かる](https://atmarkit.itmedia.co.jp/ait/articles/2112/01/news005.html)
- [大きめのCSVファイルの集計速度について比較してみた](https://zenn.dev/okkez/articles/cde47f659e9a10)

## 変換(Transform)
- このPhaseでは、データベースに変換するロジックを実装なります。このため、どれだけ並列して動かせるかがポイントなります。（=ETLのクラスタ機能が、このphaseで一番大切になります）

```mermaid
flowchart TD
    subgraph ETLツール
        Controller(呼出元/Controller) 
    end
    subgraph 変換/Transform
        Controller -->|処理依頼| Process1((プロセス1))
        Controller -->|処理依頼| Process2((プロセス2))
        Controller -->|処理依頼| Process3((プロセス3))
        Controller -->|処理依頼| Process4((プロセス4))
        Controller -->|処理依頼| Process5((プロセス5))
        Controller -->|処理依頼| Process6((プロセス6))
    end
```

### 余談
- 私は、Fargateのworker処理で実装したことがあります。SQSのキュー数でオートスケールする形を取りましたが、振り返ってみると、ETLのクラスタ機能を使うのが一番シンプルかと思います。

## 書き出し(Load)
- このPhaseでは、DBへの書き込みの高速化を考えます。
- ここでのポイントはDBコネクション数とロックになります。

### DBコネクション数
- DBのmasterは通常１つなので、DBのコネクション数の限界はシステムの限界になりますので、注意が必要です。
- 一方で、AuroraMySQLのServerlessを選ぶと、コネクション数（vCPU）をカジュアルに変更できます。
    - Aurora Capacity Unit（ACU）を変更することでvCPUを変更でき、コネクション数も上げられます。
        ![スクリーンショット 2025-02-20 9.35.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/89696/5b6d3681-1683-4cc2-ae34-b92a0038742a.png)

    - ServerlessDatabaseCapacityで現在のACU数を計測（監視）できます。
        ![スクリーンショット 2025-02-20 9.40.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/89696/eb3c7081-3d75-4ff9-816d-d60e2ea3b061.png)


### ポイント
- テクニック的にはプロセス単位でコネクション数を制限（制御）できれば、AuroraMySQLのコネクション数を超えることなく、DBに書き込めると思います。

#### 参考
- [Aurora Serverless v2 の最大接続数 (max_connections)を計算しよう](https://blog.serverworks.co.jp/aurora-serverless-v2-max_connections)
- [Amazon Aurora Serverless v2で監視すべきCloudWatchメトリクス](https://dev.classmethod.jp/articles/monitor-amazon-aurora-serverless-v2-metrics/)


### ロック
- マルチプロセスから大量に更新しようとすると`Lock wait timeout exceeded`が発生することがあります。原因は、意図しないテーブルロックが発生するからです。

#### 例
以下のテーブルがあったとします。
```sql
CREATE TABLE users (
    id INT NOT NULL,
    name VARCHAR(255),
    PRIMARY KEY (id)
) ENGINE=InnoDB;
```
以下のように、インデックスを貼っていないカラムで検索すると、フルスキャンが発生しテーブルロックになります。
```sql
BEGIN;
SELECT * FROM users WHERE name = 'Alice' FOR UPDATE;
```

#### ポイント
- ETLツールを使っているなら、ETLツールに任せましょう。
- ETLツールを使っていない場合は、RDBのトランザクションを諦めて、Redis/Valkeyを使ったMutexロックをお勧めします。

#### 参考
- [MySQLのテーブルロック・行ロックについて勘違いしていたこと](https://qiita.com/wadakatu/items/9deb535b5037f05348c0)

#### Embulk
- テンポラリテーブルを作ってから目的のテーブルにWriteすることで、ロックを回避しています。

```mermaid
flowchart LR
    Process1((プロセス1)) --> |テーブル＆レコード作成|Temp1[(仮テーブル
    1)]
    Process2((プロセス2)) --> |テーブル＆レコード作成|Temp2[(仮テーブル
    2)]
    Process3((プロセス3)) --> |テーブル＆レコード作成|Temp3[(仮テーブル
    3)]
    Temp1 --> |書き込み|Table[(目的のテーブル)]
    Temp2 --> |書き込み|Table
    Temp3 --> |書き込み|Table
```

※ うる覚えなので、間違ってたらごめんなさい m(_ _)m