データ分析

BigQuery からメールへのエクスポートの自動化

2020年5月21日

Google Cloud Japan Team

※この投稿は米国時間 2020 年 5 月 9 日に、Google Cloud blog に投稿されたものの抄訳です。

データアクセシビリティとデータ分析は、データから価値ある情報を得るために欠かせません。BigQuery ではデータの表示方法がさまざまありますが、一般的な方法の 1 つが、スケジュールに基づいてクエリ結果をメールとしてエクスポートすることです。これにより、エンドユーザーは最新のクエリ結果へのリンクが記載されたメールを受け取ることができます。ビジネスプロセスに関する日次の統計情報、サイトに関する月次のデータの概要、週次のビジネスレビューを求めている方など、どなたにも最適なソリューションです。クエリが何であれ、情報を必要とする利害関係者は、関連する分析情報を得るためにメールからデータに簡単にアクセスできます。

この投稿では、BigQuery からメールに結果をエクスポートする操作を簡単に自動化する方法について説明します。

結果メールの設計上の検討事項

設計上の重要な検討事項は、データのサイズと複雑さです。メールの添付ファイルや BigQuery 内からの大規模クエリのエクスポートにはサイズ制限があることにご注意ください。クエリ結果が 1 GB を超える場合、BigQuery は結果を複数のテーブルに出力します。その結果、複数の CSV 添付ファイルをメールで送信する必要があります。

G Suite へのアクセス権を持っている場合は、スケジュール設定された Apps Script を使用してこの送信を行うことができます。Apps Script は BigQuery API を使用してクエリを実行し、結果を Google スプレッドシートにエクスポートします。スクリプトの時間ベースのトリガーは、スケジュール設定された間隔でデータを更新します。これで、Gmail サービスを利用して、スプレッドシートへのリンクが記載されたメールを簡単に送信できます。

これは G Suite を使用した方法です。より一般的なソリューションについては、BigQuery からメールへのエクスポートを自動化する主要なソリューションとして、Google Cloud を使用することをおすすめします。この方法では、Google Cloud のいくつかのプロダクトと、メールを送信するための SendGrid API を使用します。手順は次のとおりです。

BigQuery の結果をメールにエクスポートするための自動プロセス

BigQuery の結果をメールにエクスポートするための自動プロセスを作成する方法を、手順とアーキテクチャ図からご説明します。

1. Cloud Functions コードを実行するトリガーとなる Pub/Sub トピックを作成します。

2. エクスポート用の BigQuery データセットと Cloud Storage バケットを設定します。

3. クエリの実行、結果のエクスポート、メールの送信を行うコードを使用して Cloud 関数を作成します。

4. Pub/Sub トピックに関連付けられた Cloud Scheduler ジョブを作成し、スケジュールに基づいて関数を自動的に実行します。

このプロセスのアーキテクチャは次のとおりです。

https://storage.googleapis.com/gweb-cloudblog-publish/images/Automating_BigQuery_architecture.max-800x800.jpg

クリックして拡大

このアーキテクチャでは次のことが確認できます。

Cloud Scheduler: Cloud Scheduler ジョブは、Pub/Sub トピックを呼び出して、メールの定期的なエクスポートをスケジューリングします。
Pub/Sub: Pub/Sub トピックが Cloud 関数をトリガーします。
Cloud Functions: Cloud Functions は Pub/Sub トピックをサブスクライブし、BigQuery と Cloud Storage API を呼び出すコードを実行します。
BigQuery: BigQuery API はクエリ結果を生成してテーブルに保存し、結果を CSV として Cloud Storage にエクスポートします。
Cloud Storage: Cloud Storage バケットは CSV ファイルを保存します。Cloud Storage API は、ユーザーにメールで送信される CSV の署名付き URL を生成します。

最後に、SendGrid API が、署名付き URL へのリンクを含むメールを指定の宛先に送信します。

メールのエクスポートを開始する

このプロセスを開始する際に、データの保存とメールの送信に関連する 1 回限りの設定の手順がいくつかあります。まず、エクスポートごとに作成されたテーブルをホストする BigQuery データセットを作成します。たとえば毎日メールを受信する場合、このデータセットには「daily_export_${TIMESTAMP}」などの命名規則を使用した日次のエクスポートごとのテーブルがあります。このデータセットはサイズがすぐに大きくなる可能性があるため、デフォルトのテーブル有効期限を設定することをおすすめします。このようにすることで、古いデータを保持しているテーブルを削除できます。

次に、BigQuery からエクスポートされた CSV ファイルをホストする Cloud Storage バケットを作成します。データセットの有効期限と同様に、バケットのライフサイクル管理の構成では、「Age」条件で定義された期間後に、CSV を自動的に削除したり、ファイルを別のストレージクラスに移動したりできます。

最後の設定手順では、SendGrid API へのアクセス権を構成します。そのためには、アカウントを作成して SendGrid API キーを生成します。これにより、Cloud Function が Email API で認証され、メールを送信できます。SendGrid の無料枠の料金は、最初の 30 日間は 1 日あたり 40,000 件のメッセージに適用され、その後は終了日なしで 1 日あたり 100 件のメッセージに適用されます（API の実装については、次のセクションでご説明します）。

実装の詳細

サービスアカウントの作成

SendGrid API を認証するには、サービスアカウントを作成する必要があります。Cloud Functions の署名済み認証情報を生成するには、サービスアカウントにサービスアカウントトークン作成者のロールが必要です。また、BigQuery と Storage のアクションを実行するためのアクセス権も必要となりますが、このために BigQuery 管理者のロールとストレージオブジェクト管理者のロールを追加します。次のサンプルコードは、前述のロールを持つサービスアカウントを作成します。

読み込んでいます...

Cloud Functions コードの記述

このソリューションを構築するには、Python クライアントライブラリを使用して Google BigQuery と Cloud Storage API を呼び出します。適切なサービスアカウントの認証情報を使用してクライアントライブラリをインスタンス化し、正しい認証を行って必要なタスクを実行します。メインスクリプトが Cloud Functions で実行される場合、認証情報はデフォルトでアプリケーションのデフォルト認証情報になります。ローカルで実行される場合、認証情報は環境変数 GOOGLE_APPLICATION_CREDENTIALS によって提供されるサービスアカウントの鍵ファイルを使用します。次のサンプルコードは認証情報の作成を示しています。

読み込んでいます...

ここで、メイン関数でクライアントライブラリをインスタンス化します。

読み込んでいます...

BigQuery と Cloud Storage クライアントライブラリを使用すると、テーブルの作成、そのテーブルへのクエリ結果の出力、Cloud Storage への CSV のテーブルデータのエクスポートができます。

次に、バケットに保存されている CSV ファイルの署名付き URL を生成します。このプロセスには、リンクにアクセスできる期間を示す有効期限の設定が含まれます。有効期限については、受信者が古いデータにアクセスできないように、メール間の時間差を設定する必要があります。認証のために、関数 ID は関数の現在の ID（関数を実行するサービスアカウント）を取得します。iam.Signer() は、そのサービスアカウントにリクエストを送信して、generate_signed_url() 関数を認証するための OAuth 認証情報を生成します。

読み込んでいます...

# Generate a v4 signed URL for downloading a blob.
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(csv_name)
    signing_credentials = None
    if os.getenv("IS_LOCAL") is None:
        signer = iam.Signer(request=requests.Request(), credentials=credentials(), service_account_email=os.getenv("FUNCTION_IDENTITY"),)
        # Create Token-based service account credentials for signing
        signing_credentials = service_account.IDTokenCredentials(
            signer=signer,
            token_uri="https://www.googleapis.com/oauth2/v4/token",
            target_audience="",
            service_account_email=os.getenv("FUNCTION_IDENTITY"),
        )
    url = blob.generate_signed_url(
        version="v4",
        # This URL is valid for 24 hours, until the next email
        expiration=datetime.timedelta(hours=24),
        # Allow GET requests using this URL.
        method="GET",
        # Signing credentials; if None falls back to client credentials
        credentials=signing_credentials,
    )

SendGrid API を使用してメールを送信するには、生成したトークンを使用してウェブ API の SendGrid 実装ガイドに沿って操作します。次のサンプルをご参照ください。

読み込んでいます...

上記のように、SendGrid API キーには Cloud Functions の環境変数としてアクセスできます。API キーをより安全に保存するために、Cloud Key Management Service でキーを暗号化して保存することもできます。

パイプラインのデプロイ

パイプラインを構築するには、Pub/Sub トピックを作成し、前のセクションのコードを使用して Cloud 関数をデプロイしたうえで、パイプラインをトリガーするように Cloud Scheduler を構成します。このサンプルコードは、「main.py」が Cloud 関数のコードを保持していると仮定して、パイプラインをローカルにデプロイする方法を示しています。

読み込んでいます...