MatillionでBigQueryのテーブルをGoogle Cloud Storage (GCS) に出力する方法

どうも、最近荒野行動をやっていて、開始早々に射線を確保しているスナイパーを発見、「スナイパーは、懐に入られると、弱い...ッ！！」と丸腰で果敢にインファイトを仕掛けるも、近接戦闘も得意とするタイプだったらしく、見事返り討ちに遭いボコボコにされてしまった、分析官の岡部です。

本記事はMatillion for BigQueryのTipsです。
BigQueryユーザーの方以外はご容赦ください。

また、Matillionとは？？という方は記事下部で補足をしているので、まずはそちらからご覧いただけたらと思います。

それではいきましょう。

BiqQueryのテーブルをGCSに出力する
- CSV形式で出力
- parquet形式で出力
補足：Matillionとは？？

BiqQueryのテーブルをGCSに出力する

唐突ですが、BigQueryに入ってるデータをGCSに出力したいこと、あるかと思います。
ワンショットで欲しければBQのコンソール画面からエクスポート機能を使ってやればいいのですが、システムに組み込みたいともなるとそうもいきません。
例えば、

BQテーブルをGCSに出力→外部サービスのAPIを叩いてそのファイルを呼び出し何らかの処理をする→返ってきた結果を再度GCSに格納

みたいなことを定期実行してやりたいような時です。
上のような処理を実現するため、MatillionからBQのテーブルをGCSに出力するにはどうすればいいのでしょう？
以下でその詳細を説明していきます。

CSV形式で出力

BQからデータを持ち出す処理をするため、Orchestration Jobを使います。

出力形式がCSVの場合は、素直にConnecters/Unload以下のCloud Storage Unloadコンポーネントを使いましょう。

f:id:gri-blog:20210612213937p:plain

BQ側の設定

Project：プロジェクト名
Dataset：データセット名
Table：テーブル名

GCS側（出力ファイル）の設定

Google Storage URL Location：ファイル出力先のパス
Output Object Name：出力ファイル名
Format：出力形式。今は「CSV」
Include Header：ヘッダーを含めるかどうか
Compression：圧縮形式
Delimiter：デリミター。今は「,」

これだけ設定すれば完了です。実行すると指定した場所にファイルが出力されているかと思います。

一つ注意点としては、BQのデータセットとGCSのバケットは同一リージョンに配置してください。現状、リージョンをまたいだデータの移動はできない仕様になっております。

f:id:gri-blog:20210612213828p:plain — 領域が違うとエラーになります

parquet形式で出力

簡単です。先程のFormatをCSVからPARQUETにするだけです。
、、、と言いたいところなのですが、そうは問屋がおろしません。
と言うのもparquet形式での出力は2021/2/16にリリースされたばかりのβ版であり、現にBQコンソール上のエクスポート機能も、parquet形式での出力にはまだ対応していないのです。
当然、Cloud Storage Unloadコンポーネントもparquet形式には対応していません。

【参考】 www.niandc.co.jp

ではどうするか？？
python コンポーネントを使います。以下詳細です。

まずはこちらの記事のように、 Bash Scriptコンポーネントで必要なパッケージをインストールしましょう。

f:id:gri-blog:20210612214218p:plain

# pipのアップグレード
python3 -m pip install --upgrade pip --user
# pandas, forecastflow, gcs sdkをインストール
python3 -m pip install google-cloud-storage --user
python3 -m pip install google-cloud-bigquery --user -U

お次はpython コンポーネントを使います。
これはpythonのコードを実行できる、いわば何でも屋みたいなコンポーネントです。
こいつを使ってBQからGCSにファイル出力していきましょう。
詳細は割愛させていただきますが、以下のように記述してください。
（BUCKET_NAME, PREFIX, BQ_PROJECT, BQ_DATASET, BQ_TABLEはご自身の環境に合わせて適宜Job Variablesに設定をお願いします。）

f:id:gri-blog:20210612214326p:plain

import google.cloud.storage
from google.cloud import bigquery
client = bigquery.Client()
storage_client = google.cloud.storage.Client()

# 出力先のGCSパス
destination_uri = f'gs://{BUCKET_NAME}/{PREFIX}/test_parquet.parquet'
# 出力したいBQテーブル
table_str = f'{BQ_PROJECT}.{BQ_DATASET}.{BQ_TABLE}'

# 出力形式の指定
bq_ext_job_config = bigquery.ExtractJobConfig(
    compression=bigquery.Compression.SNAPPY,
    destination_format=bigquery.DestinationFormat.PARQUET
)

# テーブル出力
extract_job = client.extract_table(
    table_str,
    destination_uri,
    # Location must match that of the source table.
    location='asia-northeast1',
    job_config=bq_ext_job_config
)  # API request

# 完了まで待つ
extract_job.result()