Cloud Storage から BigQuery に csv を流したい

nbstsh

外部のデータを csv に export した上で BigQuery に流したい。
ざっと調べたところ、csv ファイルを Cloud Storage に置いて、BigQuery に流すのが良さげなので具体的な方法をメモしていく。

いくつか方法があるみたいなので、

シンプルでサクッと実現できる
一連の処理を自動化できる
できる限りコストを抑えたい

この辺の要件を満たす方法を探す。

nbstsh

Storage にテスト用の csv アップロードしとく

まずテスト用の csv を生成する

mockaroo という便利なサイトを発見したので使ってみる

https://www.mockaroo.com/

nbstsh

Cloud Storage Bucket 作っとく

nbstsh

先ほど作成したテスト用 csv アップロード

nbstsh

Loading CSV data from Cloud Storage

公式 Doc 貼っとく

nbstsh

一旦 Console から手動でやってみる

まず、Console から手動で BigQuery に CloudStorage の csv をアップロードしてみる

nbstsh

dataset 作る

nbstsh

Storage の csv から Table 作成

"Create Table" から入力フォームを開く。

Source

Google Cloud Storage
bucket は先ほど作成したものを選択
file は先ほどアップロードしたものを選択
File Format は CSV

Destination

dataset は先ほど作成したものを選択
Table Name は適当に入力

nbstsh

ちゃんと反映されてる！想像よりはるかに簡単

nbstsh

これだと自動化できないので次は Node.js でやってみる

nbstsh

TypeScript で Storage から BigQuery に csv を流す

とりあえず ts の setup

npm install --save-dev typescript tsx

npx tsx --init

package.json

  "scripts": {
    "dev": "tsx index.ts"
  }

index.ts

console.log('Hello World!');

実行

npm run dev
// Hello World!

準備完了

nbstsh

install dependencies

@google-cloud/bigquery, @google-cloud/storage をインストール

npm install --save @google-cloud/bigquery @google-cloud/storage

nbstsh

サンプルコードをもとに実装

わかりやすいように table 名を users_from_ts にしておく

index.ts

import { BigQuery, JobLoadMetadata } from '@google-cloud/bigquery';
import { Storage } from '@google-cloud/storage';

const bigquery = new BigQuery();
const storage = new Storage();

// Storage の情報
const BUCKET_NAME = 'gcp-bigquery-sample';
const CSV_FILE_NAME = 'users.csv';
// BigQuery の情報
const DATASET_ID = 'cloud_storage_sample';
const TABLE_ID = 'users_from_ts';

async function loadCSVFromGCS() {
  // Imports a GCS file into a table with manually defined schema.

  // Configure the load job. For full list of options, see:
  // https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#JobConfigurationLoad
  const metadata: JobLoadMetadata = {
    sourceFormat: 'CSV',
    skipLeadingRows: 1,
    autodetect: true,
  };

  // Load data from a Google Cloud Storage file into the table
  const [job] = await bigquery
    .dataset(DATASET_ID)
    .table(TABLE_ID)
    .load(storage.bucket(BUCKET_NAME).file(CSV_FILE_NAME), metadata);

  // load() waits for the job to finish
  console.log(`Job ${job.id} completed.`);

  // Check the job's status for errors
  const errors = job.status?.errors;
  if (errors && errors.length > 0) {
    throw errors;
  }
}

loadCSVFromGCS().catch(console.error);

nbstsh

Application Default Credentials

BigQuery と Storage で認証を通すために、Application Default Credentials をセットアップする。
今回は手元で実行できればいいので、gcloud コマンドでセットアップする。

gcloud auth application-default login

nbstsh

実行

script を実行

npm run dev

確認

users_from_ts table が作成されていることを確認！

nbstsh

あとはこの処理を Cloud Scheduler に乗っけてあげれば、定期的に csv を BigQuery に流せるようになるな

nbstsh

BigQuery Data Transfer Service を使ってみる

The BigQuery Data Transfer Service for Cloud Storage allows you to schedule recurring data loads from Cloud Storage to BigQuery.

Cloud Storage から BigQuery への定期的なデータロードをスケジュールすることができるサービス

これを使えば自前でコード書かずに定期読み込みが実現できそう

nbstsh

試してみる

BigQuery > Data transfers から "CREATE TRANSFER" を選択

初回の場合は、BigQuery Data Transfer API を有効化する

新規 Transfer を作成していく

nbstsh

設定値を入力していく

nbstsh

エラー出た

Error code 5 : Not found: Table playground-387d6:cloud_storage_sample.users_transfered

Unlike individual loads of data from Cloud Storage to BigQuery, for ongoing transfers you must create the destination table before setting up the transfer. For CSV and JSON files, you must also define the table schema in advance. BigQuery cannot create the table as part of the recurring data transfer process.

事前に table を作成しておく必要があるみたい。
尚且つ、schema も定義しておかないといけないとのこと。

nbstsh