Cloud FunctionsでFirestoreからStorageにCSVをエクスポートする方法（Python3.7編）

はじめに

株式会社ニューズ・ラインのバックエンドエンジニア（インフラもほんのりやってます）のemuraです。最近、UberEatsデビューしてマクドナルドを注文しましたが、セットのポテトが入っていませんでした。マクドナルドの店員さんお疲れのようですね、ご苦労様です。

さて、弊社で開発中のサービスに機械学習を利用したレコメンドエンジンを導入すべく、GCP（Google Cloud Platform 以降GCPとする）を学習しながら、日々構築を進めています。今回はその中で利用しているFirestoreのデータをCloud Functionsを使ってStorage（Google Cloud Storage 以降Storageとする）にCSVでエクスポートする方法を説明します。

事前に必要なこと

GCPのプロジェクトに対する課金（Blazeプラン）が有効になっていること
Firestoreにデータがあること
StorageにCSVをエクスポートするバケットの用意

FirestoreからStorageへCSVをエクスポートする

FirestoreのデータをStorageにエクスポートする方法はいくつかありますが、今回はCloud Functionsを利用します。Cloud Functionsの管理画面を開き、新規で関数を作成していきます。（関数の基本設定は省く）

ランタイムの選択

「Pytthon 3.7」を選択
※プログラム言語を指します。

エントリポイントの設定

今回は「main」を設定
※プログラムの開始メソッドの名前です。

requirements.txtの設定

下記のように書きます。

# Function dependencies, for example:
# package>=version
google-cloud-firestore==1.2.0
google-cloud-storage==1.23.0

main.pyの設定

下記のように書きます。
【】の部分は自身のGCPの環境の設定値に変更してください。

import base64
import json
import csv
from io import StringIO
from google.cloud import firestore
from google.cloud import storage

# Cloud Storageのパスを定義
BUCKET_NAME = '【CSVをエクスポートするStorageのバケット名】'

def main(event, context):
    # Firestoreに接続
    db = firestore.Client()
    # 第一階層のコレクションを取得
    main_ref = db.collection('【第一階層のコレクション名】')
    # CSV作成用のリスト作成
    list = []
    # 第一階層コレクションのドキュメントを取得
    for doc_ref in main_ref.list_documents():
        # 第二階層コレクションを取得
        sub_ref = doc_ref.collection('【第二階層のコレクション名】')
        # 第二階層コレクションのドキュメントを取得
        for sub_doc in sub_ref.stream():
            # 第一階層コレクションのドキュメントIDと第二階層コレクションのフィールド1をリストに追加
            list.append([doc_ref.id, sub_doc.get('【第二階層ドキュメントのフィールド1のフィールド名】')])

    # CSV出力準備
    si = StringIO()
    # CSVの値にダブルクオートを付与
    cw = csv.writer(si, quoting=csv.QUOTE_ALL)
    # CSVに出力
    cw.writerows(list)
    # CSVファイルをCloud Storageへアップロード
    upload_blob(BUCKET_NAME, si.getvalue(), '【ファイル名】.csv')

# Cloud Storageへのアップロード処理
def upload_blob(bucket_name, source_file_name, destination_blob_name):
    # Cloud Storageに接続
    storage_client = storage.Client()
    # バケット情報を取得
    bucket = storage_client.get_bucket(bucket_name)
    # CSVファイルを設定
    blob = bucket.blob(destination_blob_name)
    # CSVファイルをアップロード
    blob.upload_from_string(data=source_file_name, content_type='text/csv')

関数のデプロイ

作成した関数をデプロイし、エクスポート先に設定したStorageのバケット配下に指定したデータのCSVファイルがエクスポートされていればOKです。

今回苦労した点

下記のように本当に情報が少なく実装にかなり時間がかかりました。

requirements.txtの書き方が分からなかった（公式含め、情報が少ない）
Firestoreデータのサブコレクションのデータの取り方が分からなかった（公式含め、情報が少ない）（main.pyの第一階層、第二階層のfor文の回し方のコード参照）
Cloud FunctionsでのPython 3.7での実装方法で分からない部分が多かった（公式含め、情報が少ない）

おわりに

今回はCloud Functionsの手動実行でのFirestoreデータのCSVエクスポートを説明しましたが、Cloud Scheduler、Pub/Subと組み合わせることによって、自動で定期的にエクスポートをすることもできます。次回以降はそのあたりの設定方法についても触れていきたいと思います。