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上...(続く)

以上.

1. やりたいこと

2. 予備知識の復習

Cloud Storage

Cloud Functions

BigQuery

3. やってみる

Cloud Storageバケットの用意

Cloud Functionsの設定

BigQueryの設定

4. 最後に

コメント