はじめに
BigQuery の Data Transfer Serviceを利用すると、Google Cloud Storage に配置した CSV ファイルを、スケジュールまたは手動実行によって BigQuery へ継続的に取り込めます。
本記事では、UTF-8 形式の CSV ファイルを BigQuery ネイティブテーブルへ取り込む基本的な手順に加え、次の内容を検証します。
- UTF-8とShift_JISの取り扱い
- 取り込み時間パーティション分割テーブルへの転送
-
CHANGES関数を利用した変更履歴の取得
Data Transfer Service の転送設定には、取り込み日時や転送元ファイル名などを任意の監査列として追加する機能はありません。一方で、取り込み時刻を基準にデータをパーティションへ格納したり、BigQuery の変更履歴機能を有効化して、ロードされたデータの変更を確認したりすることは可能です。
検証用 CSV ファイルの作成
はじめに、Data Transfer Service で取り込む検証用のCSVファイルを作成します。
今回は、日本語データを含む次のファイルを UTF-8 で保存します。
sale_id,store_id,item_id,quantity,amount,sold_at
100101,東京店,商品001,1,1010.0,2026-06-15T10:01:00+09:00
100102,東京店,商品002,2,2010.0,2026-06-15T10:02:00+09:00
100103,大阪店,商品003,3,3010.0,2026-06-15T10:03:00+09:00
100104,大阪店,商品004,4,4010.0,2026-06-15T10:04:00+09:00
100105,名古屋店,商品005,5,5010.0,2026-06-15T10:05:00+09:00
Cloud Storage への CSV ファイルの配置
作成したCSVファイルをData Transfer Serviceから参照できるように、Cloud Storage バケットへアップロードします。
転送先ネイティブテーブルの作成
CSVファイルの列名、列順、データ型に合わせて、転送先となる BigQuery ネイティブテーブルを作成します。
CREATE OR REPLACE TABLE
abc.transfer_demo_csv (
sale_id INT64,
store_id STRING,
item_id STRING,
quantity INT64,
amount NUMERIC,
sold_at TIMESTAMP
)
;
データ転送の作成
BigQueryの「データ転送」画面から、新しいデータ転送を作成します。
今回は、以下のように設定します。
| 設定項目 | 設定値 |
|---|---|
| ソースのタイプ | Google Cloud Storage |
| 転送構成名 | transfer_from_csv |
| 転送先データセット | abc |
| 宛先テーブルのタイプ | Native table |
| Destination table | transfer_demo_csv |
| Cloud Storage URI | transfer_demo_bucket/CSV/s*.csv |
| Write preference | APPEND |
| File format | CSV |
| Encoding | UTF8 |
| Field delimiter | , |
| Header rows to skip | 1 |
ネイティブテーブルへの転送結果の確認
転送を手動実行し、CSV ファイルのデータが BigQuery テーブルへ正常に追加されたことを確認します。
SELECT
*
FROM abc.transfer_demo_csv
;
Shift_JIS (Windows-31J / CP932) の取り扱い
データ転送設定の Encoding で Shift_JIS を選択することはできないです。
UTF-8の設定のまま Shift_JIS 形式の CSV ファイル取り込むと、エラーにはなりませんが、文字化けします。
そのため、Shift_JIS 形式のファイルは、Data Transfer Service で取り込む前に UTF-8 へ変換しておく必要があります。
取り込み時時間パーティションへの転送
続いて、データ転送の実行日を基準として、データを取り込み時間パーティションへ格納します。
Cloud Storage 転送では、転送の run_time を取り込み時刻として使用し、転送先テーブル名にランタイム パラメータを指定できます。
取り込み時間パーティション テーブルの作成
転送先として、日単位の取り込み時間パーティション テーブルを新たに作成します。
CREATE OR REPLACE TABLE
abc.transfer_demo_csv_partitioned (
sale_id INT64,
store_id STRING,
item_id STRING,
quantity INT64,
amount NUMERIC,
sold_at TIMESTAMP
)
PARTITION BY
_PARTITIONDATE
OPTIONS (
require_partition_filter = FALSE
)
;
ランタイム パラメータを使用した転送設定
データ転送設定の Destination table に transfer_demo_csv_partitioned${run_time|"%Y%m%d"} と設定します。
転送先テーブルをパーティション テーブルに変更した場合の挙動
最初に非パーティション テーブルを転送先に設定している場合、転送先テーブルをパーティション テーブルに変更しようとすると、以下のエラーが発生します。
Incompatible table partitioning specification. Expected partitioning specification none, but input partitioning specification is interval(type:month)
取り込み時間パーティション テーブルへの転送結果の確認
転送後、_PARTITIONDATE を参照して、データがどのパーティションへ格納されたかを確認します。
SELECT
*,
_PARTITIONDATE AS pt
FROM abc.transfer_demo_csv_partitioned
;
ロード履歴の取得
最後に、BigQuery の変更履歴機能を有効化し、Data Transfer Service によってロードされたレコードを CHANGES 関数で確認します。
ますは、対象テーブルの enable_change_history オプションを有効にします。
ALTER TABLE abc.transfer_demo_csv_partitioned
SET OPTIONS (
enable_change_history = TRUE
);
Data Transfer Service によるロードも変更履歴として取得することができました。
SELECT
*
FROM CHANGES(
TABLE abc.transfer_demo_csv_partitioned,
TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 12 HOUR),
CURRENT_TIMESTAMP()
);
転送履歴について
Data Transfer Service が内部で実行した BigQuery ジョブは、INFORMATION_SCHEMA.JOBS_BY_PROJECT で確認できます。
しかし、取り込んだファイル名は取得できませんでした。
SELECT
job.*
FROM
`region-asia-northeast1`.INFORMATION_SCHEMA.JOBS_BY_PROJECT AS job
WHERE EXISTS (
SELECT
1
FROM
UNNEST(job.labels) AS label
WHERE
label.value = 'data_transfer_service'
);
https://docs.cloud.google.com/bigquery/docs/information-schema-jobs?hl=ja