MongoDBにCSVから大量のデータを流し込む

# はじめに

業務用のアプリを作ってるとCSVファイルからデータベースに何かを入れたり更新するというのは日常的に出てくる要望です
みんな大好きなMicrosoft Excelで編集できるというのが一番のメリットでしょうか

MongoDBの場合、無加工でもいい状態ならmongoimportでもある程度はまかなえたりしますが通常何も考えずにExcelから書き出されるCSVファイルというのは文字コードがShift JISだったり文字列ゆえの日付や時刻データのハンドリング、あるいはアプリケーション固有のデータに落とし込むなどの何かしらの変換が必要で、そのためのプログラムを書くなんていうのもまたよくある話です

## 今回のお題
サンプルデータとして[なんちゃって個人情報](http://kazina.com/dummy/)さんから全フィールド入り5000件を6回生成して合体させた30000件のCSVを食わせてMongoDBに突っ込むというスクリプトを作ってみます


## CSVファイルを変換しつつ読み込む
CSVファイルの文字コードがShift JISでデータベースはUTF-8というのはよくある話です
さらにCSVファイルの中には半角カナ、全角英数字が入り乱れているのも非開発部門から来るファイルによくみられる特徴です

PHPの場合、[mb_convert_encoding()](http://php.net/manual/ja/function.mb-convert-encoding.php)で文字コードを変換してあげて、[mb_convert_kana()](http://php.net/manual/ja/function.mb-convert-kana.php)を使って全角半角の統一などの変換をすることになるのですが、ファイルが巨大な場合は一度全部ファイルの内容をメモリに読み込んで変換する方法はなるべくとりたくありません

1行ずつ読み込んで都度変換・・・というのもクオート処理もろもろで面倒くさいことこの上なくなります

PHPには[fgetcsv()](http://php.net/manual/ja/function.fgetcsv.php)という行ベースのシーケンシャルアクセスでCSVを処理するにはもってこいの関数があるのですが、マルチバイト環境では地雷原のようなものでした(fgetcsvでぐぐるとハマった先人たちの苦闘の記録がたくさんあります)

そんな苦闘の中から福音がもたらされることになります
ストリームフィルタを使って等価的にマルチバイト系の変換をしながらfgetcsv()で読み込めばいいのです

- [巨大なSJISのCSVファイルをfgetcsv関数で処理する](http://d.hatena.ne.jp/hnw/20090317)
- [Stream_Filter_Mbstringを使えばCSVの処理が捗る！！](http://nob-log.info/2012/04/17/stream_filter_mbstring/)

実装は下記にあります

- [Stream_Filter_Mbstring@Openpear](http://openpear.org/#stream_filter_mbstring)
- [xcezx/Stream_Filter_Mbstring@GitHub](https://github.com/xcezx/Stream_Filter_Mbstring)

今回はGitHubからzipとしてダウンロードしてメインのファイル直下にsrcフォルダ以下のStreamディレクトリを配置しました
composerで入れることもできるらしいのですが何やらひと手間かかる模様です

参考→ [fgetcsvを使ってSJISのCSVをロードするとき
](http://qiita.com/maimai-swap/items/b41059dfa38456e098e0)








## BulkWriteを使ってみる
MongoDBには[BulkWrite](https://docs.mongodb.com/manual/core/bulk-write-operations/)というコレクションに対してINSERT/UPDATE/DELETE/UPSERTをバッチ的に行える機能があります
今回はPHP7 + [MongoDB driver](http://php.net/manual/en/set.mongodb.php) + [mongodb/mongo-php-library](https://github.com/mongodb/mongo-php-library)でこのBulkWriteを使ってみます

mongodb-php-libraryでのBulkWriteの使い方はこちらで説明されています

参考→ [BulkWrite](http://mongodb.github.io/mongo-php-library/classes/collection/#bulkwrite)



## 実際に動かす

愚直にINSERTしていくバージョンとBulkWriteバージョンのPHPソースと必要なライブラリをそろえるためのcomposer.jsonを準備します
今回は進捗状況を表示するために[guiguiboy/PHP-CLI-Progress-Bar](https://github.com/guiguiboy/PHP-CLI-Progress-Bar)を利用しました

参考→[PHPのCLIスクリプトでプログレスバー](http://qiita.com/khirose/items/8b5c52c9a5a1f8a276ba)

```json:composer.json
{
    "mongodb/mongodb": "^1.0.0",
    "guiguiboy/php-cli-progress-bar": "dev-master"
}
```

### 通常INSERT版

```php:linewriter.php
<?php

require_once dirname(__FILE__) . '/vendor/autoload.php';
require_once dirname(__FILE__) . '/Stream/Filter/Mbstring.php';

stream_filter_register('convert.mbstring.*','Stream_Filter_Mbstring');

$mongoClient = new MongoDB\Client('mongodb://localhost:27017');
$db = $mongoClient->selectDatabase('nanchatte');
$collection = $db->selectCollection('persons');
$collection->drop();

sleep(1);

$fp = fopen('nanchatte.csv', 'r');
for($lines=0; fgets($fp); $lines ++);
rewind($fp);
print "$lines lines.\n";
$bar = new \ProgressBar\Manager(0, $lines);


stream_filter_append($fp,'convert.mbstring.encoding.SJIS-win:UTF-8');
stream_filter_append($fp,'convert.mbstring.kana.KVas:UTF-8');

$time_start = microtime(true);
while ($columns = fgetcsv($fp, 4096)){
    $collection->insertOne([
        'name' => $columns[0],
        'kana' => $columns[1],
        'email' => $columns[2],
        'gender' => $columns[3],
        'age' => $columns[4],
        'birth' => $columns[5],
        'bride' => $columns[6],
        'blood' => $columns[7],
        'prefecture' => $columns[8],
        'tel1' => $columns[9],
        'tel2' => $columns[10],
        'carrier' => $columns[11],
        'curry' => $columns[12]
    ]);
    $bar->advance();
}
$time_end = microtime(true);
$time = $time_end - $time_start;

echo $time . " seconds\n";
```

### BulkWrite版

```php:bulkwriter.php
<?php

require_once dirname(__FILE__) . '/vendor/autoload.php';
require_once dirname(__FILE__) . '/Stream/Filter/Mbstring.php';

stream_filter_register('convert.mbstring.*','Stream_Filter_Mbstring');

$mongoClient = new MongoDB\Client('mongodb://localhost:27017');
$db = $mongoClient->selectDatabase('nanchatte');
$collection = $db->selectCollection('persons');
$collection->drop();

sleep(1);

$fp = fopen('nanchatte.csv', 'r');
for($lines=0; fgets($fp); $lines ++);
rewind($fp);
print "$lines lines.\n";
$bar = new \ProgressBar\Manager(0, $lines);


stream_filter_append($fp,'convert.mbstring.encoding.SJIS-win:UTF-8');
stream_filter_append($fp,'convert.mbstring.kana.KVas:UTF-8');

$time_start = microtime(true);
$bulks = [];
while ($columns = fgetcsv($fp, 4096)){
    $bulks[] = [
        'insertOne' => [[
            'name' => $columns[0],
            'kana' => $columns[1],
            'email' => $columns[2],
            'gender' => $columns[3],
            'age' => $columns[4],
            'birth' => $columns[5],
            'bride' => $columns[6],
            'blood' => $columns[7],
            'prefecture' => $columns[8],
            'tel1' => $columns[9],
            'tel2' => $columns[10],
            'carrier' => $columns[11],
            'curry' => $columns[12]
        ]]
    ];
    // 1000件ずつ書き込んでおいたほうが安全らしい
    if(count($bulks) > 999){
        $collection->bulkWrite($bulks);
        $bulks = [];
    }
    $bar->advance();
}

// 残りのBulkwriteを処理する
if(count($bulks) > 0){
    $collection->bulkWrite($bulks);
}

$time_end = microtime(true);
$time = $time_end - $time_start;

echo $time . " seconds\n";
```





## 同一スペックのマシンで3回ほど流してみた結果(単位は秒)
|                  |    通常INSERT     |    Bulk Write    |
|:-----------------|------------------:|-------------------:|
| 1回目|40.45|24.52|
| 2回目|41.43|26.05|
| 3回目|40.09|25.15|

実務でも日次バッチで巨大な20万行ほどのCSVファイル食わせる処理を書いていたのですが、もともとのCSVファイルの処理方法のマズさもあって(たぶんメモリ全読み込みのせいでスワップしてたと思われる)えらく時間がかかっていたのですが、ストリームフィルタ読みとBulkWrite書き込みに変更したところ劇的に速くなりました