GCP - Cloud Functions with Pythonについて調査した

仕様

ランタイム

• 実行環境：Python3.7 on Ubuntu 18.04

ソースコードの構造

デプロイ時にアップロードされるファイルは構造化された条件に従っている必要がある。

• 例1

.
└── main.py

• 例2

.
├── main.py
└── requirements.txt

• 例3

.
├── main.py
└── requirements.txt
└── mylocalpackage/
    ├── __init__.py
    └── myscript.py

Cloud Functions の命名

プロイ時に name プロパティが設定される。
関数の名前は識別子として使用されるため、リージョン内で一意でなければならない。

依存関係の指定

公開パッケージの利用

requirements.txt内に依存パッケージを記述するとデプロイ時にpip経由でインストールしてくれる。
下記パッケージはデプロイ時に関数と一緒に自動的にインストールされる。
関数のコードでこれらのパッケージのいずれかを使用している場合、次のバージョンを requirements.txt ファイルに含めることをおすすめします。

aiohttp==3.6.2
async-timeout==3.0.1
attrs==19.3.0
cachetools==4.1.1
certifi==2020.6.20
chardet==3.0.4
click==6.7
Flask==1.0.2
google-api-core==1.21.0
google-api-python-client==1.9.3
google-auth==1.18.0
google-auth-httplib2==0.0.3
google-cloud-core==0.28.1
google-cloud-trace==0.19.0
googleapis-common-protos==1.52.0
grpcio==1.30.0
httplib2==0.18.1
idna==2.8
itsdangerous==0.24
Jinja2==2.10
MarkupSafe==1.0
multidict==4.7.6
opencensus==0.1.6
pip==18.0
protobuf==3.12.2
pyasn1==0.4.8
pyasn1-modules==0.2.8
pytz==2020.1
PyYAML==5.3.1
requests==2.21.0
rsa==4.6
setuptools==40.2.0
six==1.15.0
uritemplate==3.0.1
urllib3==1.24.3
Werkzeug==0.14.1
wheel==0.31.1
wrapt==1.10.11
yarl==1.4.2

Cloud Functions の関数のタイプ

Cloud Functions には、HTTP 関数とバックグラウンド関数の 2 つのタイプがある。

HTTP 関数

標準的な HTTP リクエストから HTTP 関数を実行します。この HTTP リクエストはレスポンスを待ち、GET、PUT、POST、DELETE、OPTIONS など、通常の HTTP リクエスト メソッドを処理します。Cloud Functions を使用する際は、TLS 証明書が自動的にプロビジョニングされるため、すべての HTTP 関数を、セキュアな接続を使用して呼び出すことができます。(HTTP 関数の作成)

sample.py

from flask import escape

def hello_http(request):
    """HTTP Cloud Function.
    Args:
        request (flask.Request): The request object.
        <http://flask.pocoo.org/docs/1.0/api/#flask.Request>
    Returns:
        The response text, or any set of values that can be turned into a
        Response object using `make_response`
        <http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>.
    """
    request_json = request.get_json(silent=True)
    request_args = request.args

    if request_json and 'name' in request_json:
        name = request_json['name']
    elif request_args and 'name' in request_args:
        name = request_args['name']
    else:
        name = 'World'
    return 'Hello {}!'.format(escape(name))

バックグラウンド関数

バックグラウンド関数を使用すると、Pub/Sub トピックのメッセージや Cloud Storage バケットの変更など、Cloud インフラストラクチャのイベントを処理できる。(バックグラウンド関数の作成)

sample.py

def hello_pubsub(event, context):
    """Background Cloud Function to be triggered by Pub/Sub.
    Args:
         event (dict):  The dictionary with data specific to this type of
         event. The `data` field contains the PubsubMessage message. The
         `attributes` field will contain custom attributes if there are any.
         context (google.cloud.functions.Context): The Cloud Functions event
         metadata. The `event_id` field contains the Pub/Sub message ID. The
         `timestamp` field contains the publish time.
    """
    import base64

    print("""This Function was triggered by messageId {} published at {}
    """.format(context.event_id, context.timestamp))

    if 'data' in event:
        name = base64.b64decode(event['data']).decode('utf-8')
    else:
        name = 'World'
    print('Hello {}!'.format(name))

ローカルマシンからのデプロイ

gcloud ツールによるデプロイ

gcloud コマンドライン ツールで gcloud functions deploy コマンドを実行して、関数のコードを含むディレクトリから関数をデプロイする。

引数	説明
NAME	デプロイする Cloud Functions の関数の登録名。
--entry-point ENTRY-POINT	ソースコード内の関数またはクラスの名前。
--runtime RUNTIME	pythonであればpython37またはpython38を指定する。
TRIGGER	関数のトリガーの種類を指定する。
--stage-bucket=STAGE_BUCKET	ソースコードを保存するバケット名を指定。

NAMEをソースコード内に含まれないカスタム文字列を指定する場合、実行するソースコードの関数は--entry-point ENTRY-POINTで指定する。
--runtimeは2回目以降のデプロイ時は省略可能。

--stage-bucketで保存先を指定する場合は毎回指定する必要がある。また、このフラグを使用する場合はバケットへの書き込み権限が必要になる。

TRIGGERについて

トリガー種別	値
HTTP	--trigger-http
Google Cloud Pub/Sub	--trigger-topic TOPIC_NAME
GCSアップロードイベント	--trigger-resource YOUR_TRIGGER_BUCKET_NAME \ --trigger-event google.storage.object.finalize

デプロイコマンド

gcloud functions deploy NAME --entry-point NAME --runtime RUNTIME TRIGGER [FLAGS...]

クイックスタート

Cloud Storage にファイルをアップロードして関数をトリガーする

サンプルシステムの概要

GCSへファイルがアップロードされたら、ファイル情報をログに出力し、さらにBigQueryへデータを挿入するところまで実装してみる。

準備

APIを有効にする

GCPシステム構成

• GCS バケット名 : sample-cloudfunctions
• BigQuery　データセット . テーブル : samples.test

Cloud Storage バケットを作成する

構文 : gsutil mb gs://YOUR_TRIGGER_BUCKET_NAME

gsutil mb gs://sample-cloudfunctions
# バケットがバージョニング対応でないことを確認
gsutil versioning set off gs://sample-cloudfunctions 

サンプルコード

Google Cloud Functions サンプル

• ファイル構成

.
├── main.py
└── requirements.txt

requirements.txt

google-cloud-bigquery

main.py

# coding: utf-8
from google.cloud import bigquery


def load_bigquery(data, context):
    """Background Cloud Function to be triggered by Cloud Storage.
         Load To Bigquery

      Args:
          data (dict): The Cloud Functions event payload.
          context (google.cloud.functions.Context): Metadata of triggering event.
      Returns:
          None; the output is written to Stackdriver Logging
      """
    print('Event ID: {}'.format(context.event_id))
    print('Event type: {}'.format(context.event_type))
    print('Bucket: {}'.format(data['bucket']))
    print('File: {}'.format(data['name']))
    print('Metageneration: {}'.format(data['metageneration']))
    print('Created: {}'.format(data['timeCreated']))
    print('Updated: {}'.format(data['updated']))

    client = bigquery.Client()

    # データセットをセット
    dataset_id = 'samples'
    dataset_ref = client.dataset(dataset_id)
    table_name = 'test'

    job_config = bigquery.LoadJobConfig()
    # スキーマ定義
    job_config.schema = [
        bigquery.SchemaField("name", "STRING"),
        bigquery.SchemaField("age", "INTEGER"),
    ]
    # 開始行はスキップ
    job_config.skip_leading_rows = 1
    # フォーマットを指定
    job_config.source_format = bigquery.SourceFormat.CSV
    # GSCのリソースURI
    uri = "gs://{}/{}".format(data['bucket'], data['name'])
    # API リクエスト
    load_job = client.load_table_from_uri(
        uri, dataset_ref.table(table_name), job_config=job_config
    )
    print("Starting job {}".format(load_job.job_id))
    # 結果を表示
    load_job.result()  # Waits for table load to complete.
    print("Job finished.")

    # テーブルからデータを取得　GCSからのデータ取り込み確認
    destination_table = client.get_table(dataset_ref.table(table_name))
    print("Loaded {} rows.".format(destination_table.num_rows))

デプロイ

gcloud functions deploy load_bigquery \
--runtime python38 \
--trigger-resource sample-cloudfunctions \
--trigger-event google.storage.object.finalize

テスト

• 予め用意していたファイルをアップロード

gsutil cp sample.csv gs://cloudfunctions-sample

• ログを確認 構文 : gcloud functions logs read [FUNCTION_NAME]

gcloud functions logs read load_bigquery --limit 15

D      load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.105  Function execution started
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Event ID: 1384453759070888
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Event type: google.storage.object.finalize
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Bucket: sample-cloudfunctions
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  File: sample.csv
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Metageneration: 1
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Created: 2020-07-29T05:43:17.582Z
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:19.117  Updated: 2020-07-29T05:43:17.582Z
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:20.537  Starting job 3b748f9e-ae61-42bc-817b-1b2bdb2c3d8a
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:24.975  Job finished.
       load_bigquery  g2dfwptfywjx  2020-07-29 05:43:25.438  Loaded 3 rows.
D      load_bigquery  g2dfwptfywjx  2020-07-29 05:43:25.441  Function execution took 6337 ms, finished with status: 'ok'

ログを確認するとLoaded 3 rows.と出力されており、予め用意しておいたデータがBigQueryへロードされたことが確認できた。