データエンジニアのモリヤです！

Airflowを使用してBigQueryのテーブルのデータをCloud Storageにエキスポートしてみた記録記事になります。

今回試してみた環境

  • Airflow : 3.1.3
  • Mac OS（Intel）
  • Docker : 24.0.2
  • Docker Compose : 2.18.1（2.1.4以降である必要があるそうです）
  • Docker Desktop
  • VSCode

※無料枠の範囲内で完結するはずですが、料金が発生する可能性があります。そこの責任は持てませんのでご了承ください。

※DockerでAirflowの環境構築されている前提の記事になります。環境構築の仕方は下記の記事を参考にしてみてください。

Dockerを使用したAirflowのローカル環境構築【Airflow】
Dockerを使用したAirflowのローカル環境構築【Airflow】この記事ではDockerを使ったAirflowのローカル環境での構築方法を説明しています。手軽に構築できるので、これからAirflowを勉強する方や試してみたいという方はぜひご覧いただき参考にしていただければと思います。...

手順

テーブル作成

BigQueryに下記クエリでサンプルテーブルを作成します。

CREATE TABLE `your_project.test_airflow.bigquery_to_gcs_operator_sample_table` (
  id STRING,
  date DATE
);

INSERT INTO `your_project.test_airflow.bigquery_to_gcs_operator_sample_table` (id, date)
VALUES
  ('a1',      DATE '2025-11-30'),
  ('b23',     DATE '2025-11-30'),
  ('abc',     DATE '2025-11-30'),
  ('9x',      DATE '2025-11-30'),
  ('5c9c',    DATE '2025-11-30'),
  ('dc9bc',   DATE '2025-12-01'),
  ('k2',      DATE '2025-12-01'),
  ('z81',     DATE '2025-12-01'),
  ('p9q',     DATE '2025-12-01'),
  ('7ab',     DATE '2025-12-01'),
  ('m3',      DATE '2025-12-02'),
  ('x9c2',    DATE '2025-12-02'),
  ('qwe',     DATE '2025-12-02'),
  ('8z',      DATE '2025-12-02'),
  ('c91',     DATE '2025-12-02'),
  ('a7d',     DATE '2025-12-03'),
  ('9',       DATE '2025-12-03'),
  ('bc12',    DATE '2025-12-03'),
  ('zz9',     DATE '2025-12-03'),
  ('1a2b3',   DATE '2025-12-03');

GCSバケットを作成

Cloud Storageのトップ画面にて、「バケットを作成」をクリックします。

バケット名を設定して、「続行」をクリックします。

ロケーションタイプを「Region」、「us-east1」を選択して「続行」をクリックします。

そのあとはデフォルトの設定のまま「続行」をクリックしていき、最後に「作成」をクリックします。

下のような注意が出ると思うので、デフォルトのまま「確認」をクリックします。

これでバケットが作成されるはずです。

Dagの作成

dagsフォルダに下記のファイルを作成します。

/dags/bigquery_to_gcs_csv.py
from datetime import timedelta

from airflow.models import Variable
from airflow.providers.google.cloud.transfers.bigquery_to_gcs import BigQueryToGCSOperator
from airflow.sdk import DAG

BQ_PROJECT = Variable.get("BQ_PROJECT")
BQ_DATASET = Variable.get("BQ_DATASET")
BQ_TABLE_ID = "bigquery_to_gcs_operator_sample_table"
GC_BUCKET = Variable.get("GC_BUCKET")

with DAG(
    "bigquery_to_gcs_csv",
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
    },
    description="BigQuery to GCS DAG",
    tags=["test"],
) as dag:

    bigquery_to_gcs = BigQueryToGCSOperator(
        task_id="bigquery_to_gcs_export",
        gcp_conn_id="google_cloud_default",
        source_project_dataset_table=f"{BQ_PROJECT}.{BQ_DATASET}.{BQ_TABLE_ID}",
        destination_cloud_storage_uris=[f"gs://{GC_BUCKET}/bigquery_to_gcs_operator_sample_table.csv.gz"],
        export_format="CSV",
        compression="GZIP",
    )

    bigquery_to_gcs

Airflowを起動して下記のようにDagが追加されていれば良さそうです。

変数の設定

Airflowでグローバルに使用できる変数が設定できます（参考）。その機能を使って、Dagで利用している「BQ_PROJECT」と「BQ_DATASET」、「GC_BUCKET」を設定します。

画面の左のメニューから「Admin>Variables」をクリックします。

「Add Variable」をクリックします。（自分は既に変数を設定しているため、下の画像に表示されています）

下のような画面になるので

下記のように設定します

  • テーブルを作成したGoogle Cloudのプロジェクト名
    • Key：BQ_PROJECT
    • Value：your_project
  • データセット名
    • Key：BQ_DATASET
    • Value：test_airflow
  • 作成したバケット名
    • Key：GC_BUCKET
    • Value：bigquery_to_gcs_operator_sample_bucket

Google Cloudの認証情報をコンテナ内にコピーする

ローカルで今回利用しているGoogle Cloudに認証して、その情報をDockerコンテナにコピーして、コンテナ内からGoogle Cloudに接続できるようにします。

まず、ローカルで下記のコマンドを実行します。

gcloud auth application-default login

下のような画面になると思うので、該当のGoogleアカウントを選択し、その後の画面で「続行」をクリックします。

下の画像のようになればOKです。

すると、ローカルの「$HOME/.config/gcloud」に「application_default_credentials.json」というファイルが作成されているはずです。

次に、上記のjsonファイルをDockerコンテナ内にマウントします。

docker-compose.yamlファイルの77行目と83行目を追加します。

docker-compose.yaml
    AIRFLOW_CONFIG: '/opt/airflow/config/airflow.cfg'
    GOOGLE_APPLICATION_CREDENTIALS: /tmp/keys/adc.json
  volumes:
    - ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags
    - ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs
    - ${AIRFLOW_PROJ_DIR:-.}/config:/opt/airflow/config
    - ${AIRFLOW_PROJ_DIR:-.}/plugins:/opt/airflow/plugins
    - ~/.config/gcloud/application_default_credentials.json:/tmp/keys/adc.json:ro
  user: "${AIRFLOW_UID:-50000}:0"
  depends_on:
    &airflow-common-depends-on

Dagを実行

準備が整ったのでDagを実行してみます。

Airflowの左メニューの「Dags」をクリック、左上の「Trigger」をクリック、表示されたダイアログの「Trigger」をクリックします。

するとDagのStateがQueued>Running>Successとなります。

実際にCloud Storageのバケットを見るとデータがエキスポートされていました。

上記のファイルをダウンロードしてサンプルで作ったデータと同じデータになっているか確認します。

gzファイルを解凍すると下の画像のようになっていたので、しっかりと指定したBigQueryのテーブルのデータがエキスポートできていました。

補足

今回はcsv形式でファイルでエキスポートしましたが、他にもjsonなどでも可能のようです。

詳しくはこちらをご覧ください。

終わりに

今回試したかった本質の部分ではないのですが、コンテナ内からGoogle Cloudへの接続の仕方で少しハマったのでGoogle Cloudの認証・認可の理解も深まり、かなり勉強にもなりましぞd(￣ ￣)

参考記事

Airflow Docker Google Cloud データエンジニア