概要

データ分析の第一歩として「データ取り込みの（部分的）自動化」をしてみました！

GCSのCSVから自動でBigQueryに取り込んでくれる Functionsを作成します。
GUI・CLIでも同様の操作が行えますが、「特定のバケットに置くと自動で取り込み」を行うことで、社内やチーム内でも気軽にBigQueryを利用できるようになります。

実際に動かしてみる

1. Google Cloud Storageにバケットを作成

こちらを参考にサクッと作成
https://cloud.google.com/storage/docs/creating-buckets?hl=ja

2. BigQueryにデータセットを作成

公式ドキュメントが見やすいですね。「gcs_temporary_dataset」としました。
https://cloud.google.com/bigquery/docs/datasets?hl=ja

3. Firebase Functions(Cloud Functions)を利用してデプロイ

取り込むCSVは「{ファイル名}_YYYYMMDD.csv」で一行目がフィールド名という規則です。
「特定にバケットにオブジェクトが作成された」ことをトリガーにすることがコードで表現できます。最高！！！

4. ストレージにCSVを配置してBigQueryに取り込んでみる

東京都オープンデータカタログサイトからCSVをダウンロードしてみました。
ファイル名の最後に「_20200425」をつけるのを忘れずに。
https://catalog.data.metro.tokyo.lg.jp/dataset/t000010d0000000068/resource/c2d997db-1450-43fa-8037-ebb11ec28d4c

実際に取り込まれたデータ

ヘッダーが日本語なのでフィールド名が壊れちゃってますが一応成功。

実際のコード

firebase functionsとtypescriptを利用して作成しました。

'use strict';
import {BigQuery} from '@google-cloud/bigquery';
import {bigquery_v2} from 'googleapis';
import Schema$JobConfigurationLoad = bigquery_v2.Schema$JobConfigurationLoad;
import * as functions from 'firebase-functions';

const projectId = // GCPのプロジェクトID
const backetName = // GCSのバケット名
const datasetId = // BQのデータセットID

export const importBigqueryTemporaryTable = functions
  .runWith({timeoutSeconds: 300, memory: '1GB'})
  .storage.bucket(`${backetName}`)
  .object()
  .onFinalize(async (object) => {
    try {
      const name = object.name!;

      // ファイル名がテーブル名のイメージ
      // hoge_YYYYMMDD以外の名前は許さない
      const matched = name.match(/(.*)_\d{8}/);
      if (!matched) {
        console.log('filename not match.');
        return false;
      }

      const tableId = matched[0];

      const bigquery = new BigQuery();
      const configuration: Schema$JobConfigurationLoad = {
        destinationTable: {
          datasetId: datasetId,
          projectId: `${projectId}`,
          tableId: `${tableId}`,
        },
        sourceFormat: 'CSV',
        sourceUris: [`gs://${object.bucket}/${name}`],
        writeDisposition: 'WRITE_TRUNCATE',
        autodetect: true,
        encoding: 'UTF-8',
        skipLeadingRows: 1, // ヘッダーはカラム定義のイメージ
      };

      await bigquery.createJob({
        configuration: {load: configuration},
      });

      console.log('success to import bigquery temporary table.', tableId);
      return Promise.resolve('success');
    } catch (e) {
      console.log('Failed to import temporary table.', e);
      return Promise.reject(e);
    }
  });

まとめ

たったこれだけのコードで、「特定のバケットに置くとBigQueryに取り込み」を自動化することができました。CSVのフォーマットが正しくないときには動作しないなどの問題はありますが、最小限の動作は行えると思います。
BigQueryのデータセットの設定でデフォルトの保存期限を設定すれば、データの消し忘れもなくなります。

質問・ご指摘お待ちしております！