Cloud Functionsで、GCSにアップロードしたCSVデータをBigQueryに読み込む

AWSしか触ったことのない私が、最近GCP(Google Cloud Platform)を使う機会があったのでメモします。Cloud Functionsのチュートリアルとしてご覧ください。

1. やりたいこと
2. 予備知識の復習
3. やってみる
4. 最後に

1. やりたいこと

Cloud Storageにデータをアップロード
データアップロードをトリガーとしてCloud Functions起動
Cloud Functionsにより、データをBigQueryにインポート

2. 予備知識の復習

一般的なクラウドサービスについては、↓の記事にも書いています。

知っていて当然？XaaSで打線を組んでみた

NishipyXaaSで打線組んでみた。異議は、Twitter(@iamnishipy)までどうぞ！！1. はじめに巷にあふれるXaaS。X as a Serviceの略です。ここで言うXは、たぶん数学とかによく出てくるあの...(続く)

Cloud Storage

オブジェクトストレージのクラウドサービスであり、AWSで言う所のS3です。
詳細は、下記リンクを参照ください。

Cloud Storage

Cloud Storage lets you store data with multiple redundancy options, virtually anywhere.

Cloud Functions

GCPの提供するFaaSであり、AWSで言う所のLambdaです。
イベントドリブンのサーバーレスアーキテクチャを実現できます。
詳細は、下記リンクを参照ください。

Cloud Functions | Google Cloud

Scalable pay-as-you-go Function-as-a-Service (FaaS) to run your code with zero server management.

BigQuery

データウェアハウスのクラウドサービスであり、AWSで言う所のAthenaだと思います。

BigQuery エンタープライズ向けデータウェアハウス

BigQuery は、ビッグデータから価値あるビジネス分析情報を得るために設計された、サーバーレスで費用対効果に優れたマルチクラウドデータウェアハウスです。ぜひ、無料トライアルをお試しください。

3. やってみる

Cloud Storageバケットの用意

Cloud Storageでは、S3と同様にバケットと呼ばれるデータの入れ物を作成します。
データへのアクセス権限の設定を省略するために、今回はバケットをインターネット上に公開することにします。

バケットの作成

RegionalまたはMulti-Regionalを選択して、作成します。

バケットの公開
下記の手順で、バケットをインターネットに公開します。
- 権限タブからユーザの追加を選択
- 「allUser」に対して、ストレージ閲覧ロールを付与
- 公開と表示されていることを確認

Cloud Functionsの設定

Cloud Functionsに関数をデプロイします。トリガーには、Cloud Storageのイベントを選択します。

各種設定項目を埋める

以下のように、値を入れます。

#	設定項目	値
1	名前	任意
2	割り当てられるメモリ	任意(最大2GB)
3	トリガー	Cloud Storage
4	イベントタイプ	ファイナライズ/作成
5	バケット	[前章で作成したバケット]

requirements.txtを書く

関数の実行に必要なパッケージとバージョンをrequirements.txtに書いておきます。

# Function dependencies

google-cloud-storage == 1.14.0

google-cloud-bigquery == 1.11.2

main.pyを書く

あとは、main.pyにデプロイする関数を書くだけです。今回はランタイムとして、Python3.7を利用します。
[YOUR PROJECT ID]および[YOUR DATASET ID]は既存のものを指定します。

関数の入力データとなるdataとcontextの説明は、下記リンクにあります。

Objects | Cloud Storage | Google Cloud

また、Google Cloud Client Library for Pythonのリファレンスはこちらです。

https://googleapis.github.io/google-cloud-python/latest/#

from google.cloud import storage, bigquery

from datetime import datetime

def load_data(data, context):

client = bigquery.Client()

# 'data' reference is here:

# https://cloud.google.com/storage/docs/json_api/v1/objects?hl=ja

################################

# TO DO: Set [YOUR PROJECT ID] #

################################

project_id = '[YOUR PROJECT ID]'

################################

# TO DO: Set [YOUR DATASET ID] #

################################

dataset_id = '[YOUR DATASET ID]'

bucket_name = data['bucket']

file_name = data['name']

file_ext = file_name.split('.')[-1]

if file_ext == 'csv':

uri = 'gs://' + bucket_name + '/' + file_name

file_ext = file_name.split('.')[-1]

suffix = datetime.now().strftime("_%Y%m%d%H%M%S")

table_id = 'tweets' + suffix

# Job Configuration

dataset_ref = client.dataset(dataset_id)

job_config = bigquery.LoadJobConfig(autodetect=True)

job_config.skip_leading_rows = 1

job_config.source_format = bigquery.SourceFormat.CSV

# Job Request

load_job = client.load_table_from_uri(

uri,

dataset_ref.table(table_id),

job_config=job_config

)

print('Started job {}'.format(load_job.job_id))

load_job.result() # Waits for table load to complete.

print('Job finished.')

destination_table = client.get_table(dataset_ref.table(table_id))

print('Loaded {} rows.'.format(destination_table.num_rows))

else:

print('Nothing To Do')

BigQueryの設定

main.pyの変数に定義した名前のデータセットを作っておきます。
他は特にやることはありません。

4. 最後に

今回書いた関数は、GitHubにあげています。あわせてご覧ください。

GitHub - nishipy/CloudFunctions

Contribute to nishipy/CloudFunctions development by creating an account on GitHub.

今後は、BigQueryに読み込んだデータを、GCPの他のサービスと連携させていきたいと思います。
追記：続きはこちら。

BigQueryに読み込んだTwitterデータを、Datalabで分析・可視化する

NishipyAWSしか触ったことのない私が、BigQueryに読み込んだデータを、Google Datalabを用いて可視化してみます。1. やりたいことBigQueryにデータを読み込み(前回やった)BIgQuery上...(続く)

以上.