Google Cloud Platformに分析基盤を構築

はじめまして、GMOあおぞらネット銀行でエンジニアをしているT.Kです。
普段は当社CTO直下部署で、新規技術調査や新規サービスなどの先行調査・検討などを行っています。
初めてのはてなブログとなりますが、複数回でGoogle Cloud Platform(以下、GCP)に構築した分析基盤に関してブログを書きたいと思います。
今回は、当社データセンターからBigQueryへの登録までに触れていきます。
GCP上にシステム構築を行ったのは初めての為、イマイチな部分も多くありますが、今後のより良い形に改善していきたいと考えています。

概要

全体イメージ

今回のブログでは、上記イメージ図のオンプレ⇒DWH部分にフォーカスしていきます。
初期要件としては、『連携は1日1回〜2回程度、合計約50MBのファイルを個別に受信し、共通DWHへ登録したい』となります。
データ活用部分やセキュリティを意識した開発環境構築などに関しては、次回以降のブログで紹介できればと考えています。

情報基盤要件と方針

• データ連携頻度は1日に数回
  汎用性と運用効率の面からスケジューラは利用しない
  バッチ式で対応
• 将来的に連携データ量(ファイル数)は増加想定
  運用負荷軽減の為、コード内に連携データ固有の処理を記述しない
• 連携データを利用した後続プロジェクトが存在し、将来的には増加想定
  データ取込処理は5分以内で完了する
• 当社セキュリティ基準をクリアする環境構築 (※次回ブログ予定)

構築の流れ

オンプレからDWHまでのサービス構成イメージ

1. Cloud Storageの準備

バケット作成 & 各種設定

• バケットを作成にはお金は掛からない為、用途別に複数作成
• Cloud FunctionからGCSへ処理が行われるケースでは、"アクセス制御"を"きめ細かい管理"にしないとエラーが発生したので、一旦"きめ細かい管理"設定
• 不要ファイルはライフサイクルで対応

バケット用途	アクセス制御	ライフサイクル	備考
受信用バケット	きめ細かい管理	【削除】最終更新後1日	Cloud Functionからデータ読込
一時保存用バケット	均一	【削除】最終更新後1日
backup用バケット	きめ細かい管理	【削除】最終更新後60日	Cloud Functionからデータ書込
設定用バケット	均一

• "アクセス制御"はバケット作成時に選択
  

• "ライフサイクル"はバケット作成後、バケット詳細から設定

• 1日後に削除するライフサイクル設定例
  

2 外部ファイル準備

今回の仕組みでは、2つのCloud Function処理で利用する、設定ファイルとmerge用SQLが対象となります。
連携データ(ファイル)固有の情報を外部ファイルに切出した為、外部ファイル更新のみで、連携データの更新が可能な仕組みとなっています。

設定ファイル

各Cloud Functionで利用する設定や連携データ固有情報を記述したファイル

▶ property file sample

{
  "settingBucket" : "_setting", 
  "common": {
    "dc_file_prefix" : "XXX_",
    "debug_mode" : 0
  },
  "firstStep": {
    "output": {
      "bucket": "_temp_working",
      "path": "first_step"
    },
    "backup": {
      "bucket": "_backup",
      "path": "original"
    }
  },
  "secondStep": {
    "bigquery": {
      "dataset": "dwh",
      "bucket": "_setting"
    },
    "pubsub": {
      "status": 0 ,
      "projectid": "-monitoring",
      "topicid": "_topic",
      "target" : [
        { "name": "XXX.csv", "message": "daily list", "comment": "XXX" },
        { "name": "XXX2.csv", "message": "monthly list", "comment": "XXX"}
      ]
    }
  },
    "acceptTables" : [
      { "name": "xxx.csv", "mergeQueryPath": "query/xxx.txt", "comment": "取込対象情報を記述" },
    ]
}

merge文ファイル

Cloud Function内で実行するSQLファイル

▶ merge SQL sample

MERGE `{}` target USING `{}` stg 
 ON target.ORIGINAL_TBL_ID = stg.TEMP_TBL_ID 
WHEN MATCHED THEN 
 UPDATE SET
  SAMPLE_NAME = stg.SAMPLE_NAME
  , DELETED_FLG = stg.DELETED_FLG as string
  , CREATED_DATETIME = TIMESTAMP_ADD(stg.CREATED_DATETIME, INTERVAL -9 HOUR)
  , CREATED = stg.CREATED
  , UPDATED_DATETIME = TIMESTAMP_ADD(stg.UPDATED_DATETIME, INTERVAL -9 HOUR)
  , UPDATED = stg.UPDATED
  , ACTIVED = stg.ACTIVED as string
  , GCP_PROCESS_DATETIME = CURRENT_TIMESTAMP() 
WHEN NOT MATCHED THEN 
INSERT(ORIGINAL_TBL_ID, SAMPLE_NAME, DELETED_FLG, CREATED_DATETIME, CREATED, UPDATED_DATETIME, UPDATED, ACTIVED, GCP_PROCESS_DATETIME) 
VALUES(
  ORIGINAL_TBL_ID
  , SAMPLE_NAME
  , DELETED_FLG as string
  , TIMESTAMP_ADD(CREATED_DATETIME, INTERVAL -9 HOUR)
  , CREATED
  , TIMESTAMP_ADD(UPDATED_DATETIME, INTERVAL -9 HOUR)
  , UPDATED
  , ACTIVED as string
  , CURRENT_TIMESTAMP()
) 

3. 第一弾処理(Cloud Function)の作成

幾つかにポイントを絞って触れていきたいと思います。

▶ main.py

from google.cloud import storage    # GCS利用の為
from zipfile import ZipFile         # zipファイル解凍の為
from zipfile import is_zipfile      # zipファイル解凍の為
import os                           # 環境変数取得の為
import io                           # 解凍ファイル展開の為
import json                         # 設定ファイル読込の為
import datetime                     # バックアップ用フォルダ生成の為

property = ""

def first_step(event, context):
    try:
        print(logJSON("INFO", "【FIRST STEP】 処理開始"))

        #プロパティファイル読込
        global property
        property = getProperty()

        if property is not None:
            #ファイル名とバケット名を引数から取得
            inputBucket = event['bucket']
            inputFile = event['name']

            #受信ファイルをbackupへコピー
            cophyFileToBackupBucket(inputBucket, inputFile)

            #zipファイル解凍処理
            zipextract(inputBucket, inputFile) 

        print(logJSON("INFO", "【FIRST STEP】 処理完了"))
    except Exception as e:
        print(logJSON("ERROR", "【FIRST STEP】 エラー発生： " + e.message))
        raise

#受信ファイルをバックアップ　バケットへコピー
def cophyFileToBackupBucket(backetName, fileName):
    #GCSクライアント
    storageClient = storage.Client()

    #GCSバケット設定
    bucket = storageClient.get_bucket(backetName)
    backupBucket = storageClient.get_bucket(os.environ.get('SYSTEM_ENV') + property["firstStep"]["backup"]["bucket"])

    #バックアップ用変数設定
    blob = bucket.blob(fileName)
    today = getDate()

    #バックアップ作成処理
    new_blob = bucket.copy_blob(blob, backupBucket, new_name=property["firstStep"]["backup"]["path"] + '/' + today + '/' + fileName)
    new_blob.acl.save(blob.acl)

    print(logJSON("INFO", '- 【FIRST STEP】【BACKUP処理】 コピー処理成功： gs://' + property["firstStep"]["backup"]["bucket"] + '/' + property["firstStep"]["backup"]["path"]  + '/' + today + '/' + fileName))

#zipファイルを解凍して、指定バケットにアップ
def zipextract(bucketname, zipfilename_with_path):
    print(logJSON("INFO", "- 【FIRST STEP】【zip解凍処理】 処理開始"))

    #GCSクライアント
    storage_client = storage.Client()
    #GCSバケット設定
    bucket = storage_client.get_bucket(bucketname)
    #outbucket = storage_client.get_bucket(property["outputBucket"])
    outbucket = storage_client.get_bucket(os.environ.get('SYSTEM_ENV') + property["firstStep"]["output"]["bucket"])
    #受信ファイルパス設定
    destination_blob_pathname = zipfilename_with_path
    
    blob = bucket.blob(destination_blob_pathname)
    zipbytes = io.BytesIO(blob.download_as_string())

    if is_zipfile(zipbytes):
        with ZipFile(zipbytes, 'r') as myzip:
            for contentfilename in myzip.namelist():
                #チェック処理呼び出し
                if checkTables(contentfilename):
                    contentfile = myzip.read(contentfilename)
                    blob = outbucket.blob(property["firstStep"]["output"]["path"] + "/" + contentfilename)
                    blob.upload_from_string(contentfile)

                    print(logJSON("INFO", "-- 【FIRST STEP】【zip解凍処理】 対象ファイルを指定先に出力： " + property["firstStep"]["output"]["path"] + "/" + contentfilename))
    else:
        print(logJSON("WARNING", "-- 【FIRST STEP】【zip解凍処理】 受信ファイルがzip形式ではない為、処理を中断： " + zipfilename_with_path))

    print(logJSON("INFO", "- 【FIRST STEP】【zip解凍処理】 処理完了"))

#取込対象ファイル可否チェック
def checkTables(fileName):
    #許可対象TBL情報
    for accept_table in property["acceptTables"]:
        if fileName == property["common"]["dc_file_prefix"] + accept_table["name"]:
            # macでファイル圧縮時のゴミファイル除外…orz
            if "MACOSX" not in fileName:
                print(logJSON("INFO", "-- 【FIRST STEP】【対象TBLチェック】 取込対象である事を確認： " + fileName))
                return True

    #取り込み対象外
    print(logJSON("WARNING", "-- 【FIRST STEP】【対象TBLチェック】 取込対象外の為、処理を中断： " + fileName))
    return False

#設定ファイル読込
def getProperty():
    storage_client = storage.Client()
    #設定ファイルのパス情報を設定
    bucket = storage_client.get_bucket(os.environ.get('SYSTEM_ENV')  + "_common_setting")

    if bucket.blob("import_dc_data/property.json").exists(storage_client):
        try:
            blob = bucket.blob("import_dc_data/property.json")

            #設定ファイルをjson形式読込
            property_file = json.loads(blob.download_as_string())

            return property_file
        except Exception as e:
            print(logJSON("ERROR", "- 【FIRST STEP】【設定ファイル】 設定ファイルに不備がある為、処理を中断"))
            return None
    else:
        print(logJSON("ERROR", "- 【FIRST STEP】【設定ファイル】 設定ファイルが存在しない為、処理を中断"))
        return None

#実行日取得
def getDate():
    dtNow = datetime.datetime.now()
    return dtNow.strftime('%Y%m%d')

#ログレベル設定
def logJSON(logLevel, message):
    return json.dumps({
        "severity": logLevel,
        "message": message,
    })

▶ requirements.txt

google-cloud-storage

ⅰ. jsonファイルの読込

• json形式のファイルはライブラリを利用する事で簡単に読み込む事が可能

# ライブラリの読込
import json  

# ファイル内容読取
blob = bucket.blob("property.json") 

# 変数にjson形式で読取
property_file = json.loads(blob.download_as_string())

ⅱ. Google Cloud Storage間のファイルコピー

• 送信先バケットに指定フォルダが無くても、自動作成
• コピー元とコピー先バケットのアクセス制御が『きめ細かい管理』設定でないとエラー発生
• Cloud Functionが利用するサービスアカウントへ適切な権限を付与

# GCSライブラリ読込
from google.cloud import storage    

# コピー元バケット
bucket = storageClient.get_bucket(backetName)
# コピー先バケット
backupBucket = storageClient.get_bucket(backetName)

# 保存先パス
new_blob = bucket.copy_blob(blob, backupBucket, new_name= "バケットパス+日付とか" + '/' + "ファイル名")
# 保存
new_blob.acl.save(blob.acl)

ⅲ. zip解凍処理

• 圧縮ファイル解凍時、ファイル内容は一旦メモリに保存されるので、メモリ設定は特に注意
  Cloud Functionの処理自体にも一定量メモリが必要の為、多少余裕が必要
  
• 現在Cloud Functionは最大8GBまで"割り当てられるメモリ"を設定可能だが、もし8GBを超える圧縮ファイルを取扱うケースでは、別構成を検討

# zipライブラリ読込
from zipfile import ZipFile         
from zipfile import is_zipfile

# GCS上の対象ファイルを設定
blob = bucket.blob(destination_blob_pathname)
# ファイル読込
zipbytes = io.BytesIO(blob.download_as_string())
# zip形式可否チェック
if is_zipfile(zipbytes):
    # zipファイル解凍
    with ZipFile(zipbytes, 'r') as myzip:
        # 解凍ファイルを個別に処理
        for contentfilename in myzip.namelist():
            # 解凍ファイル読込
            contentfile = myzip.read(contentfilename)
            # 保存先バケットパスを設定
            blob = outbucket.blob("ファイルパス" + "/" + contentfilename)
            # 解凍ファイル保存
            blob.upload_from_string(contentfile)

ⅳ. デプロイ

• トリガー設定 ※左画像
  1. "トリガーのタイプ"に『Cloud Storage』を選択
  2. "Event Tyep"に『選択したバケット内のファイル(最終処理 / 作成)』を選択
  3. "バケット"に"起動条件バケット"を選択

• メモリ設定 + α ※右画像
  1. 解凍ファイルサイズ等に合わせて、適切な値を選択
  2. 想定処理時間等に合わせて、タイムアウトを設定(最大540秒)
  3. 必要に応じて、環境変数設定

トリガー設定　　　　　　　　　　　　　　メモリ設定 + α　

• デプロイ対象コード
  • コードは"インラインエディタ"で直接記述する事や"Cloud Source Repositories"からソースを展開する事が可能
    "Cloud Source Repositories"のソース編集はcloud shellなどを利用する必要があります

デプロイ対象コード

4. 第二弾処理(Cloud Function)の作成

こちらもポイントを絞って触れてきます

▶ main.py

from google.cloud import storage
from googleapiclient.discovery import build
from google.cloud import bigquery
from google.cloud import pubsub_v1
import os
import json

property = ""

def second_step(event, context):
    print(logJSON("INFO", "【SECOND STEP】 データ登録処理開始"))
    print(event)

    #プロパティファイル読込
    global property
    property = getProperty()

    if property is not None:
        file_name = os.path.splitext(os.path.basename(event['name']))[0]

        uri = "gs://" + str(event['bucket']) + "/" + str(event['name'])
        table_id = os.environ.get('GCP_PROJECT') + "." + property["secondStep"]["bigquery"]["dataset"] + "." + file_name.replace(property["common"]["dc_file_prefix"], '')
        table_id_temp = '{}_temp'.format(table_id)
        
        query = getSettingValues(file_name + ".csv", "mergeQueryPath")
        if query is not None:
            query = query.format(table_id, table_id_temp).replace('\n' , '')
            print(logJSON("DEBUG", "- 【SECOND STEP】 マージクエリ: " + query))

            client = bigquery.Client()
            try:
                # load data to temp table
                job_config = bigquery.LoadJobConfig()

                #一時TBL用のスキーマ情報読込
                job_config.schema = getSchema(file_name.replace(property["common"]["dc_file_prefix"], ''))
                
                job_config.skip_leading_rows = 1
                job_config.source_format = bigquery.SourceFormat.CSV
                job_config.allow_quoted_newlines = True
                job_config.write_disposition = 'WRITE_TRUNCATE'
                job_config.autodetect = True            
                table = bigquery.Table(table_id_temp)
                job = client.load_table_from_uri(uri, table, job_config=job_config)

                job.result()
                print(logJSON("INFO", "- 【SECOND STEP】 一時TBL作成成功 - job id:{} / table:{}".format(job.job_id, table.table_id)))

                # マージ文実行
                job = client.query(query)
                job.result()
                print(logJSON("INFO", "- 【SECOND STEP】 正規TBL更新成功    - job id:{} / table:{}".format(job.job_id, table_id)))

                # pubsub発行
                if bool(property["secondStep"]["pubsub"]["status"]):
                    for target_file in property["secondStep"]["pubsub"]["target"]:
                        if file_name + ".csv" == property["common"]["dc_file_prefix"] + target_file["name"]:
                            #pubsub発行処理呼出
                            setPubSubTopic(target_file)

            except Exception as e:
                print(logJSON("ERROR", "- 【SECOND STEP】 マージ処理内でエラー発生： " + e.message))
            finally:
                # デバッグモードON(1)の時は一時TBLを削除しない
                if not bool(property["common"]["debug_mode"]):
                    # 一時TBL削除
                    client.delete_table(table_id_temp, not_found_ok=True)
                    print(logJSON("INFO", "- 【SECOND STEP】 一時TBL削除処理成功 - job id:{} / table:{}".format(job.job_id, table.table_id)))
        else:
            print(logJSON("ERROR", "- 【SECOND STEP】 - queryが存在しないため、処理を中断"))

    print(logJSON("INFO", "【SECOND STEP】 データ登録処理終了"))


#設定ファイル読込
def getProperty():
    storage_client = storage.Client()
    #設定ファイルのパス情報を設定
    bucket = storage_client.get_bucket(os.environ.get('SYSTEM_ENV')  + "_common_setting")

    if bucket.blob("import_dc_data/property.json").exists(storage_client):
        try:
            blob = bucket.blob("import_dc_data/property.json")

            #設定ファイルをjson形式読込
            property_file = json.loads(blob.download_as_string())

            return property_file
        except Exception as e:
            print(logJSON("ERROR", "- 【SECOND STEP】【設定ファイル】 設定ファイルに不備がある為、処理を中断: "  + e.message))
            return None
    else:
        print(logJSON("ERROR", "- 【SECOND STEP】【設定ファイル】 設定ファイルが存在しない為、処理を中断"))
        return None

#一時テーブルのスキーマ情報取得
def getSchema(fileName):
    client = bigquery.Client()
    schemaList = []
    count = 0

    #BigQueryクライアント生成
    table_id = os.environ.get('GCP_PROJECT') + "." + property["secondStep"]["bigquery"]["dataset"] + "." + fileName
    tableInfo = client.get_table(table_id)

    #対象テーブルのスキーマ情報追加
    for field in tableInfo.schema:
        #共通項目”GCP_PROCESS_DATETIME”項目を除外
        if count < len(tableInfo.schema) - 1:
            schemaList.append(field)
            count = count+1

    return schemaList

#各TBL用外部情報取得
def getSettingValues(fileName, target):
    #許可対象TBL情報
    for accept_table in property["acceptTables"]:
        #print(fileName + ":" + accept_table["name"])
        if fileName == property["common"]["dc_file_prefix"] + accept_table["name"]:
            storage_client = storage.Client()
            bucket_name = os.environ.get('SYSTEM_ENV') + property["secondStep"]["bigquery"]["bucket"]
            file_name = accept_table[target]
            
            bucket = storage_client.get_bucket(bucket_name)
            blob = storage.Blob(file_name, bucket)
            content = blob.download_as_string()
            
            return content.decode('utf-8')
    #取り込み対象外
    return None

#pubsub発行処理
def setPubSubTopic(target_file):
    publisher = pubsub_v1.PublisherClient()

    # topicの指定 
    topic_path = publisher.topic_path(os.environ.get('SYSTEM_ENV') + property["secondStep"]["pubsub"]["projectid"], os.environ.get('SYSTEM_ENV') + property["secondStep"]["pubsub"]["topicid"])

    # メッセージ生成
    data = target_file["message"].encode('utf-8')

    # pub/subを発行
    future = publisher.publish(topic_path, data=data)
    print(logJSON("INFO", "- 【SECOND STEP】 PubSub登録成功: {} / ID {}".format(target_file["comment"], future.result())))

    return 'finished'

#ログレベル設定
def logJSON(logLevel, message):
    return json.dumps({
        "severity": logLevel,
        "message": message,
    })

▶ requirements.txt

google-cloud-storage
google-cloud-bigquery
google-api-python-client
requests==2.27.1
google-cloud-pubsub

ⅰ. 既存TBLのスキーマ情報取得

• 一時TBL用のスキーマ情報を読込
• 補足
  BigQueryは賢いので、スキーマ未指定でもデータ取込処理の実行は可能だが、都度取込データに依存したデータ型が選択されてしまう為、型にブレが出る事に注意

# BigQueryのライブラリ読込
from google.cloud import bigquery

# BigQueryのクライアント作成
client = bigquery.Client()
# スキーマ取得元のテーブルID設定
table_id = os.environ.get('GCP_PROJECT') + "." + property["secondStep"]["bigquery"]["dataset"] + "." + fileName
# スキーマ取得元のテーブル情報設定
tableInfo = client.get_table(table_id)

# 1項目ずつスキーマ情報を設定
for field in tableInfo.schema:
    if count < len(tableInfo.schema):
        schemaList.append(field)
        count = count+1

ⅱ. BigQueryへデータ登録

• 受信ファイルのデータを一時TBLへ登録

# BigQueryのライブラリ読込
from google.cloud import bigquery

# データ登録用コンフィグ設定
job_config = bigquery.LoadJobConfig()

# スキーマ情報読込 ※ ⅰの処理を行うfunction呼出
job_config.schema = getSchema(file_name.replace(property["common"]["dc_file_prefix"], ''))
# 取込データ1行目SKIP ※項目名等が存在する事を想定
job_config.skip_leading_rows = 1
# 取込データ形式設定
job_config.source_format = bigquery.SourceFormat.CSV
# 引用符で囲まれた改行を許可設定
job_config.allow_quoted_newlines = True
# 上書き設定 ※新規TBLへの登録なので不要かも…
job_config.write_disposition = 'WRITE_TRUNCATE'
#  一時TBLのIDを設定
table = bigquery.Table(table_id_temp)

# 作成先、IDとコンフィグ情報を設定
job = client.load_table_from_uri(uri, table, job_config=job_config)
# 一時TBL作成処理実行
job.result()

ⅲ. 外部ファイルからSQL文を読込・実行

• 既存TBLと一時TBLを利用したmerge文を実行
  各TBL用のmerge文を外部ファイルに切出しましたが、既存TBLのスキーマ情報を利用すれば自動生成する事も出来そうですね
  SQLサンプルはmerge SQL sample参照

# BigQueryのライブラリ読込
from google.cloud import bigquery
# BigQueryクライアント設定
client = bigquery.Client()

# 登録先TBL用のmerge文を外部ファイルから読込 ※functionの詳細は"第二弾 code"参照
query = getSettingValues(file_name + ".csv", "mergeQueryPath")
# 実行時に不要な改行コードを削除 ※外部ファイル上では見にくいので改行コードしている為
query = query.format(table_id, table_id_temp).replace('\n' , '')

# merge文設定
job = client.query(query)
# merge文実行
job.result()

ⅳ. デプロイ

• トリガー条件のバケット以外、第一弾処理と同様

5. BigQueryの設定

BigQueryに関しては登録用のTBLを作成しておくだけなので、適宜CRETE文などを実行して作成すればOK

結果

方針達成に関して

• スケジューラは利用しない & バッチ式で対応
  • GCSへのファイルアップをトリガーにBigQuery登録までの処理を自動化
• コード内に連携データ固有の処理を記述しない
  • 取込データ固有の内容に関しては、全て外部ファイルに切出す事で対応
    固有情報を切出した事で、取込データ追加等の対応に関しても基本的にコード改修不要
• 5分以内で完了
  • 下記、稼働後10日程の平均処理時間
    ファイルアップロードからBigQuery登録完了までの処理を約1分程度で完了。将来的なデータ増加に伴い、処理時間増加は想定されるが、ある程度余裕がある結果となった。

受信ファイルサイズ	第一弾処理時間	第二弾処理時間	処理合計時間
約60MB	15秒	46秒	61秒

その他

• 利用コストに関して
  • 下記、稼働後10日間の各サービスの平均費用額
    BigQueryの費用は他と比べ高めだが、Cloud Functionの費用がかなり安い結果となった。BigQueryの費用に関しては、merge文が影響している可能性があるので別途調査予定

Cloud Storage	Cloud Function	BigQuery
76円	39円	1,950円

まとめ

今回フォーカスした仕組みに関しては、(日々の連携ファイルサイズが小さい事もありますが)受信ファイル毎に並列実行可能な仕組みとした事で、処理時間に関しては全く問題ない結果となりました。
また、連携データ毎固有部分を外部に切り出せた事で、Cloud Functionのコードをシンプルに構成する事が出来ました。(必ずしも外部に切り出せば良い訳ではありませんが…)
利用費用に関しては、一部調査課題を残していますが、現状問題ない範囲で収まっている印象です。

今回の連携データではデータ加工処理が不要であった為、Cloud StorageとCloud Functionの組み合わせで特に問題はありませんでしたが、理想はDataflowを利用するべきだとは感じてました。将来、同様の仕組みを構築する場合にはDataflowに挑戦してみたいと思います。

参照サイト

• Google Cloud Platform本家ドキュメント
• 【Cloud Functions】Storageに置かれたGZIPファイルを自動解凍 - Qiita