Cloud Functionsで大量のデータを処理する

こんにちは。Stamp Incの[nori](https://twitter.com/1amageek)です。

巷でかなり多く、Firebaseの言葉を聞くようになりました。
[FJUG](https://firebase.asia/)を始めてからもう3,4年くらい経ってるんですね。たくさんの人に使って便利さを実感してもらって普及活動を頑張った僕としては嬉しい限りです。

僕が[開発したOSSのBallcap](https://github.com/1amageek/Ballcap-iOS/issues/28)が、[Firebase Open Source](https://firebaseopensource.com/)に掲載されるようです。👏🏻
コントリビュートしてくれた開発者のみなさんありがとうございました。


今日は、Cloud Functionsで大量のデータを処理するときのTipsを紹介します。
例えば、バッチ処理を組んでいて、数万件のデータを処理しないといけない場合とか、外部APIからとんでもサイズのデータが流れてくる場合が考えられます。

# Cloud Functionsで大量のデータを処理

結論から言うとこれも分散処理をさせることが重要です。
今回は、郵便番号データをCloudFunctionsで処理したサンプルがあるのでそれを例に説明しようと思います。

[サンプルコード](https://github.com/1amageek/jp-zipcode)
[郵便番号データ](https://www.post.japanpost.jp/zipcode/download.html)

## Cloud Functionsについて

### Cloud Functionsの種類

まずCloud Functionsの種類について説明します。Cloud Functionsは大きく二つの関数に分かれています。簡単に分けるならば、直接的に呼び出す関数と、副次的に呼び出す関数になります。
同じCloud Functionsを利用した関数であってもスケール特性に差があることを知っておくのは重要です。

| 関数 | 説明 | 特徴 |
| --- | --- | --- |
| HTTP関数 | httpプロトコルから呼び出すことができる関数。__functions.https.onCall__, __functions.https.onRequest__ が該当します。 | トラフィックを処理するために迅速にスケールアップ |  
| バックグラウンド関数 | 他のサービスからイベントトリガーによって呼び出される関数。 <br>__Cloud Firestore__<br>__Database__<br>__Remote Config__<br>__Authentication__<br>__Analytics__<br>__Crashlytics__<br>__Cloud Storage__<br>__Cloud Pub/Sub__<br>__Test Lab__<br> が該当します。　| 緩やかにスケールアップ |

### Cloud Functionsの上限

Cloud Functionsは3つ観点から制限が設定されています。
これらの上限は、実行時の話だけでなく、関数そのものの容量やデプロイに関することも含まれています。

| 上限 | 説明 |
| --- | --- |
| リソースに関する上限 | ファンクションの処理に使用できるリソースの合計量に影響します。 |
| 時間に関する上限 | 実行できる時間の長さに影響します。 | 
| レートに関する上限 | Cloud Functions API を呼び出すレートと、リソースを使用できるレートに影響します。レートの割り当ては、「一定時間あたりのリソース」と考えることができます。 | 

### Cloud Functionsのスケーラビリティ

今回は、Cloud Functionsの実行時に関する話をしようと思うので特に以下の上限について話をします。
まずは以下の上限があることを知っておいてください。

| 割り当て | 上限 | 説明 |
| --- | --- | --- |
| 関数の最大実行時間 | 540 秒 | 強制終了されるまで関数を実行できる時間の上限 | 
| 1 秒あたりの関数呼び出し数 | 1 秒あたりの関数呼び出しの回数。これを超過すると、次の割り当て期間まで関数は一時停止したままになります。 | 100 秒あたり 100,000,000 | 
| 最大関数メモリ | 2,048 MB |  関数が使用できるメモリ容量 |

## Cloud Functionsの性能を向上させる

### 実行するマシン性能をあげる

開発者が最もCloud Functionsの制限を感じるのはまずメモリでしょう。Cloud Functionsをデフォルトで利用すると128MBのメモリで実行され1分でタイムアウトします。

次のように`runWith`へオプションを渡してあげることで、性能を向上させることが可能です。
`timeoutSeconds`の上限は540、`memory`の上限は`2GB`です。

```typescript
const runtimeOpts = {
  timeoutSeconds: 300,
  memory: '1GB'
}

exports.myStorageFunction = functions
  .runWith(runtimeOpts)
  .storage
  .object()
  .onFinalize((object) = > {
    // do some complicated things that take a lot of memory and time
  });
```

また、メモリをあげるとCPUの割当も向上します。CPUとメモリの対応は次のようになっています。
2GBを超えるような処理はCloud Functionsには向いていません。

![スクリーンショット 2020-08-03 17.34.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/80287/5d192cdf-bfaf-5152-0e8c-9d3a7c11a348.png)
https://cloud.google.com/functions/pricing#compute_time


## サンプルコードでの実行例

今回はサンプルコードで次の処理を行いました。

- 郵便番号データ（ZIPデータ）をダウンロード（1.7MB）
- ZIPデータを解凍（展開後のデータ12.3MB）
- ShiftJISでエンコード
- データを一行づつ読み込み
- データベースへ保存

ZIPデータの展開までは最小メモリで大丈夫でしたが、一行づつ読み込んでデータベースへ保存するところでメモリが不足し実行できませんでした。当然ですが数十万行あるデータを通信を返して保存するので、チリも積もればですね。

> ※うまく構成すれば、この関数一つだけでも十分捌けるかもしれません。今回は次で説明するやり方を試してみたかった。

### 処理の分散化
今回はCloud Functionsから別のCloud Functionsを起動させて処理を分散化させました。
<img width="1088" alt="スクリーンショット 2020-08-03 18.12.56.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/80287/48984cc7-0fbc-be3d-f133-c9273dc21734.png">
こういった構成を__Fan-out__と呼びます。

スケジューラーで実行しているCloud Functionsと、`Pub/Sub`を使ってトピックをサブスクライブするCloud Functionsを準備しました。

前段のCloud Functions内部では、データを一行づつ読み込むところまで行い、500行に一度溜め込んんだ500行分のデータをパブリッシュするようにしました。500行の理由はCloud Firestoreのバッチ処理の上限が500だからです。

二段のCloud Functionsでは５００行のデータを`WriteBatch`に詰めてCloud Firestoreに保存する処理のみを行っています。

この構成により、前段のデータがどれだけ大きくなっても（2GBを超えないくらいであれば）保存処理によって処理が停止することはありません。

さらに話を進めるとFan-outの構成にもCloud Functionsの上限が引っかかることがありますが、このくらいの処理ではその上限に達しないので割愛します！と言うかその上限に引っかかるようならCloud Functionsのやめ時かもですね。。。
Cloud Runなどへの以降を推奨します！


### おまけ


```typescript
import * as functions from "firebase-functions"
import { PubSub } from "@google-cloud/pubsub"

const pubsub = new PubSub();

// 毎日に実行される処理
export const everyDay = functions
    .pubsub
    .schedule("* 0 * * *")
    .onRun(async (context) => {
      functions.logger.log("📅 Running scheduled task.")
      const topic = pubsub.topic("funout");
      // 何かしらの処理を100処理に分散
      [...Array(100).keys()].map(async (offsetMonth) => {
        const data = { 
           // your data 
        }
        await topic.publishMessage({ json: data })
      })
    })

// 分散化された処理
export const funout = functions
    .pubsub
    .topic("funout")
    .onPublish((message, _) => {
      const data = message.json
      // Your process
    })
```

備忘録として置いておきます！ぜひ参考にしてください。