Mongoに溜まった約1.6億の記事データをBigQueryで高速検索!

2021年12月23日

植江田和成

URLをコピーしました！

こんにちは、開発本部の植江田和成です。

今回は、WebClipping で使っている MongoDB に保存されていた約1.6億の記事データを BigQuery で検索できるようにしたことについて書きます。

WebClipping とは様々なサイトから記事をクロールし、その記事にユーザーが設定したキーワードが含まれていればクリップしたりなど、メディア露出の調査・分析などがおこえるWebアプリケーションです。

PHPでのファイル操作やファイル作成処理の速度向上、BigQuery へのデータ転送方法など色々学びがありましたので、紹介します。

WebClippingのMongoDBについて

WebClippingでは、MongoDBをクロールした記事の保存に使っています。

このMongoDBのクラスタは、Primary/Secondary2台を1シャードとし、それが3シャードの合計9つのmongodの構成となっており、それらは9台のAWS EC2インスタンス上に構築されています。

また、シャーディングのためにアプリケーションからのアクセスはmongosを用いてルーティングを行っています。
mongosについて: https://docs.mongodb.com/manual/core/sharded-cluster-query-router/

なぜMongoDBからデータ移動を行なったのか

今回は省略しますが、2021年9月に過去記事検索機能の障害が起きてしまいました。
この障害がデータ移動を行う理由を生んだ出来事です。

過去記事検索機能はWebClipping内で提供されており、検索日から1年前の記事を検索できる機能です。
過去記事検索について: https://webclipping.jp/service/pastsearch

この過去記事検索機能はElasticsearchへリクエストを行います。そのElasticsearchのインデックスに不具合が起きてしまい、MongoDBに保持されている元データから、インデックスを再作成する対応を行いました。
Elasticsearchについて: https://www.elastic.co/jp/what-is/elasticsearch

前述の新しいインデックス作成時に、MongoDBの作業をしていて気づいたことがありました。
それは、元々「MongoDBには、2年以上昔の記事を削除する」仕様があったはずでしたが、実際には5年以上の記事が保存されており、削除されていなかったことです。

MongoDB は大量のデータを長期アーカイブすることには不向きなので、昔の記事は削除することが考慮されているはずでしたが、それがなされていませんでした。

せっかく溜まった記事データを削除するだけではなく、BigQuery に移行することで、Mongo から消しつつもデータは残して活用できるのではないか? と考えました。

しかし、本件対応のために緊急的にMongoサーバーのインスタンスタイプをm4.xlarge(vCPU:4, メモリ:16)からr4.8xlarge(vCPU:32, メモリ:244)に変更しました。
この時には処理速度が遅い理由に気づかず、単純にサーバースペックをあげましたが、Mongoに適切なインデックスが貼られていなかったため、処理が遅くなっていました。
サーバー費用がかなり高くなってしまっている状態なので、コスト的にも速やかにスペックダウンする必要があります。

移動させる前に確認し気づいたこと

実際にデータを移動させる前にBigQueryへデータを入れてエラーが出ないか確認します。少量の件数で csv ファイルを作成し、bq loadコマンドを使用してBigQueryへデータを入れていきます。
bq コマンドについて: https://cloud.google.com/bigquery/docs/bq-command-line-tool?hl=j

この時点で確認できたのは、3つのエラーでした。

まず1つ目は、スキーマレスなこととは関係ない壊れ方ですが、正しく取得できなかったURLが格納されていたことです。protocol://domain/dir/file…</p>適当な文字列 のように、URL の中に </p> が含まれていたデータが存在していました。これはURLのなかに</p>があれば捨てる処理を追加して解消しました。

2つ目は、昔のデータと最近のデータでカラム数に差異があったことです。
間近のデータには存在しますが、2年ほど前のデータには存在しないカラムがありました。これは全てのキーの存在チェックを行い、ない場合nullで処理するように修正して解消しました。

３つ目は、BigQueryでint型のカラムにstring型のデータを入れようとしたエラーです。
これは当初、カラムの位置が動的なMongoDBから取得したデータにも関わらずカラムの順番を揃えていなかったため、途中のデータからズレたカラムを代入しようとしていたのが原因でした。

静的なカラム順序を持つBigQueryに対応するため、入力データのカラム順序を必ず揃えるようにして解消しました。

以上3つのエラーは解消できたので実際にデータを移動させていきます。

実際にデータを移動する

今回は以下の方法で移動させます。

データ量1.6億件分のcsvファイルを作成する。
S3へアップロードする。
S3からBigQueryへ転送する。

それぞれ何を行いどのような問題が発生したかを紹介します。

ひたすらファイルを作り続ける

試験的に生成した少量のデータをbq loadコマンドで転送した際にエラーが確認できなくなったので、1.6億件のデータをひたすらcsvファイルにしていきます。

PHPのStream Wrapperを用いて圧縮し、ディスク容量を抑えながら書き込んでいきます。
参考: https://www.php.net/manual/ja/wrappers.compression.php

圧縮したファイルは30Mbyteほどで、約600ファイル作ったところでサーバー監視アプリからディスク容量が残り20%以下であるというアラートが出ていました。圧縮しなければ1ファイルの容量は、おそらく150Mbyteほどになっていたはずなので、約120ファイル作成した時点でアラートが出てしまっていた可能性があります。

以下が初期実装の大まかなコードです。

for (たくさん繰り返す) {
    $cursor = Mongoからデータ取得();
    $data = $cursor->toArray();
    $decode = json_decode(json_encode($data), true);
    $firstData = $decode[0];
    $lastData = $decode[count($decode) - 1];
    // ファイル名に id を使いたいので、取得したデータの最初と最後を渡す
    $filename = ファイル作成処理($firstData, $lastData);
    
    // ファイルを圧縮するために compress.zlib ラッパーを付与
    $fh = fopen("compress.zlib://” . $filename, "w");
    
    // 配列のデータを一行ずつファイルへ書き込む
    foreach($data as $row) {
        $filtered["id"] = isset($row["_id"]) ? $row["_id"] : "";
        $filtered["s_id"] = isset($row["s_id"]) ? $row["s_id"] : "";
        fputcsv($fh, $filtered)
    }
}

この初期実装時で1ファイル3万件で10 ~ 15 秒ほどかかっていました。

単純計算したところ、1.6億件のデータの csv ファイルを1ファイル3万件で作成するには、約6840ファイル × 15秒 = 102,600秒 → 28.5 時間ほどかかる予想が経ちました。

これだと遅いので、改善していきます。

この実装の問題点は、配列のデータを1行ずつファイルに書き込むように実装してしまったことです。

// 配列のデータを一行ずつファイルへ書き込む
foreach($data as $row) {
    $filtered["id"] = isset($row["_id"]) ? $row["_id"] : "";
    $filtered["s_id"] = isset($row["s_id"]) ? $row["s_id"] : "";
    .
    .
    fputcsv($fh, $filtered)
}

ファイルへの書き込み頻度が多かったため、3万件でも時間がかかってしまっていました。

そこで、1行ずつ書き込むのではなくまとめてファイルへ書き込むようにします。

下記のようにコードを書き換えました。

for (たくさん繰り返す) {
    $arr = mongoからデータ取得();
    // メモリに書き込む
    $temp_memory_file_fh = fopen("php://memory", "r+");
    fputcsvで書き込む($temp_memory_file_fh,$arr);
    fseek($temp_memory_file_fh, 0);

    // ファイル名作成のため最初と最後のデータを取得
    $firstData = (array)$arr[0];
    $lastData = (array)$arr[count($arr) - 1];
    $tmpFilePath = ファイル作成処理($firstData,$lastData);
    $csvTmpFile = fopen('compress.zlib://' . $tmpFilePath, 'w');

    // 実ファイルへ書き込み
    stream_copy_to_stream($temp_memory_file_fh, $csvTmpFile);
    fclose($temp_memory_file_fh);
    fclose($csvTmpFile);
}

この改善で、1ファイル5万件に拡張しても8~10秒で処理が進み、総ファイル数も約半分の3200ファイルで済むようになり約2倍の高速化に成功しました。

また、この実装では配列データを直接ディスク上のファイルへ書き込むのではなく、一度メモリに書き込み、書き込んだ内容を一気にファイルへ書き込む手法も採用しています。

// メモリに書き込む
$tempMemoryFile = fopen("php://memory", "r+");
$temp_memory_file_fh = fputcsvで書き込む($tempMemoryFile,$arr);
.
.
// 実ファイルへ書き込み
stream_copy_to_stream($temp_memory_file_fh, $csvTmpFile);

メモリに書き込むことでディスクへの書き込み頻度を減らすことができ、処理速度を引き上げることに成功しました。

作成したファイルをS3へアップロードする

次は作成したファイルをアップロードしていきます。

なお、今回は時間やディスクの節約のためにファイル生成と同時にアップロードを実施しています。

ライブラリはAWS SDK for PHPを使用しました。
参考: https://aws.amazon.com/jp/sdk-for-php/

// 指定ディレクトリ配下にあるファイルを配列で取り出す
$fileList = glob("/path/to/old_articles/*");
if (!$fileList) {
    var_dump("ファイルが0件です。処理を終了します。");
    exit();
} else {
    var_dump("ファイル総数 : " . count($fileList));
}

$S3 = new AwsS3();
$localPath = "path/to/old_articles/";
$S3Path = "files/";

foreach ($fileList as $file) {
    $exploded = explode("/", $file);
    $filename = $exploded[count($exploded)-1];
    var_dump("filename : " . $filename);

    // 圧縮されたファイルを解凍するため compress.zlib ラッパーを付与
    $filenameWithWrapper = "compress.zlib://" . $file;
    $unzipped = fopen($filenameWithWrapper, 'r');
    fclose($unzipped);

    $uploaded = $S3->S3_upload_file($file, $S3Path . $filename);
    if ($uploaded) {
        var_dump("ファイルのアップロードに成功しました。");
        $unlinked = unlink($file);
        if ($unlinked) {
            var_dump("ファイルを削除しました : " . $file);
        } else {
            var_dump("ファイルを削除できませんでした。 : " . $file);
        }
    } else {
        var_dump("ファイルのアップロードに失敗しました。");
    }

ディスク容量を圧迫するので、アップロードができたファイルから削除していきます。
圧縮したファイル約50Mbyte × 400件ほどで、サーバー監視アプリからディスク容量が残り20%アラートが出ていました。

上記の容量不足問題を回避するため、全てのファイル生成をしてからアップロードではなく、ファイル制作とアップロードを逐次で実行することで無事に完了させました。

S3からBigQueryへ転送する

さて、ファイルも作成できアップロードも完了したのでS3からBigQueryへ転送していきます。

BigQueryのデータ転送画面から諸々設定して、実行します。
参考記事: https://cloud.google.com/bigquery-transfer/docs/s3-transfer?hl=ja

しかし、ここでエラーが発生してしまいました。レコード数約1.6億件、全3200ファイルを作成しきった後にも関わらず。

具体的には、以下のようなエラーです。

Cannot return an invalid timestamp value of 1552896131000000000 microseconds relative to the Unix epoch. The range of valid timestamp values is [0001-01-01 00:00:00, 9999-12-31 23:59:59.999999];

1万年以上未来の異常なタイムスタンプになっていました。

データを作り直す際に0001-01-01 00:00:00, 9999-12-31 23:59:59.99999の範囲を超えないデータだけを配列に格納する処理とすることで解消できました。

この問題は稀に発生するものなので、生成済みの巨大なデータでそれだけをフィルタリングして抜くことが困難でした。よって分割して、できるだけ狭いレンジの中にエラーデータが存在するファイルを見つける方法で解決することにしました。

例えばs3://articles/data/の配下に全てのファイルが存在するなら、s3://articles/data/1/2などのように分割して転送を繰り返し、できるだけ狭いレンジのなかにエラーデータが見つかるようにします。

最終的に転送画面は以下のようになってしまいました。

*S3 URI は一部簡略化されています。