以下の手順でGoogle BigQueryからPostgreSQLにテーブルをコピーします。

BigQueryからテーブルのスキーマとデータをAvro形式でexportする
本記事に記載のPythonコードにAvroファイルを渡して、PostgreSQLへのimport用SQL文を出力させる
そのSQL文をPostgreSQLで動かしてimportする

その後、BigQueryとPostgreSQLで同じSELECT文を実行して、同じ結果が返ることをざっくり確認します。

※ 場合により、テーブルの完全なコピーにならないこともありますので、ご了承ください。
　ご利用は自己責任でお願いします。

BigQueryからテーブルのスキーマとデータをAvro形式でexportする

BigQueryの「エクスポート」メニューから「GCSにエクスポート」を選択します。

「GCSのロケーション」を設定し、エクスポート形式は「Avro」を選択して保存ボタンを押します。

GCSからローカルのマシンにAvroファイルをダウンロードします。

PostgreSQLへのimport用SQL文を生成する

コマンドライン引数にAvroファイル名を渡して、以下のPythonコードを動かします。
Avroファイル名から拡張子を除いた前半部分が、PostgreSQLのテーブル名として使用されます。
出力されるSQLファイル名は、そのテーブル名に拡張子sqlをつけたものになります。

bq_avro_to_postgres_db.py


import sys
import traceback
import json
from pathlib import Path
import fastavro
from collections import deque
from psycopg2.extensions import adapt


class MyException(Exception):
    pass


class MyReader(fastavro.reader):
    def __init__(self, file):
        super().__init__(file)
        self.meta = self._header["meta"]


def escape(ss):
    return str(adapt(ss.encode("utf-8").decode("latin-1")))


def convert_to_postgres_type(avro_type, default_value, ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type):
    postgres_type = None
    default_str = None
    nullable = False

    if default_value is None:
        default_str = "NULL"

    if type(avro_type) is list:
        if "null" in avro_type:
            nullable = True
            avro_type.remove("null")
        assert len(avro_type) == 1
        avro_type = avro_type[0]

    if type(avro_type) is str:
        if avro_type == "string":
            postgres_type = "VARCHAR"
            if default_value is not None:
                default_str = escape(default_value)
        elif avro_type == "bytes":
            postgres_type = "BYTEA"
            if default_value is not None:
                pass
        elif avro_type == "long":
            postgres_type = "BIGINT"
            if default_value is not None:
                default_str = str(default_value)
        elif avro_type == "double":
            postgres_type = "DOUBLE PRECISION"
            if default_value is not None:
                default_str = str(default_value)
        elif avro_type == "boolean":
            postgres_type = "BOOLEAN"
            if default_value is not None:
                if default_value:
                    default_str = "TRUE"
                else:
                    default_str = "FALSE"
    elif type(avro_type) is dict:
        if "sqlType" in avro_type:
            sql_type = avro_type["sqlType"]
            if sql_type == "JSON":
                postgres_type = "JSON"
                if default_value is not None:
                    default_str = escape(default_value)
        elif "logicalType" in avro_type:
            logical_type = avro_type["logicalType"]
            if logical_type == "decimal":
                postgres_type = "NUMERIC"
                if default_value is not None:
                    default_str = str(default_value)
            elif logical_type in ("timestamp-millis", "timestamp-micros"):
                postgres_type = "TIMESTAMP WITH TIME ZONE"
                if default_value is not None:
                    default_str = escape(default_value.isoformat())
            elif logical_type == "date":
                postgres_type = "DATE"
                if default_value is not None:
                    default_str = escape(default_value.isoformat())
            elif logical_type in ("time-millis", "time-micros"):
                postgres_type = "TIME WITH TIME ZONE"
                if default_value is not None:
                    default_str = escape(default_value.isoformat())
            elif logical_type in ("local-timestamp-millis", "local-timestamp-micros"):
                postgres_type = "TIMESTAMP"
                if default_value is not None:
                    default_str = escape(default_value.isoformat())
        else:
            avro_child_type = avro_type["type"]
            if avro_child_type == "array":
                default_value = None
                postgres_items_type, _, _ = convert_to_postgres_type(
                    avro_type["items"], default_value, ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type)
                if type(postgres_items_type) is str:
                    postgres_type = {
                        "array": postgres_items_type + "[]",
                        "items": postgres_items_type
                    }
                else:
                    postgres_type = {
                        "array": postgres_items_type["record"] + "[]",
                        "items": postgres_items_type
                    }
            elif avro_child_type == "record":
                len_is_serial_of_postgres_record_type.append(True)
                postgres_record_type = f"{postgres_record_type_prefix}_type_{len(len_is_serial_of_postgres_record_type)}"
                postgres_fields_type = make_sql_create_type(
                    postgres_record_type, avro_type["fields"], ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type)
                postgres_type = {
                    "record": postgres_record_type,
                    "fields": postgres_fields_type
                }
    return postgres_type, default_str, nullable


def make_sql_create_type(name, avro_fields, ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type):
    ddl_stmt = ""
    postgres_fields_type = []
    for field in avro_fields:
        default_value = None
        postgres_type, _, _ = convert_to_postgres_type(
            field["type"], default_value, ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type)
        postgres_fields_type.append(postgres_type)
        if len(ddl_stmt) == 0:
            ddl_stmt = f"CREATE TYPE {name} AS (\n    "
        else:
            ddl_stmt += "\n  , "
        ddl_stmt += field["name"] + ' '
        if type(postgres_type) is str:
            ddl_stmt += postgres_type
        elif "record" in postgres_type:
            ddl_stmt += postgres_type["record"]
        else:
            ddl_stmt += postgres_type["array"]
    if 0 < len(ddl_stmt):
        ddl_stmt += "\n);\n"
        ddl_queue.append(ddl_stmt)
    return postgres_fields_type


def make_sql_create_table(name, avro_fields, ddl_queue, postgres_record_type_prefix):
    ddl_stmt = ""
    postgres_type_list = []
    len_is_serial_of_postgres_record_type = []
    for field in avro_fields:
        if "default" in field:
            default_value = field["default"]
        else:
            default_value = None
        postgres_type, default_str, nullable = convert_to_postgres_type(
            field["type"], default_value, ddl_queue, postgres_record_type_prefix, len_is_serial_of_postgres_record_type)
        postgres_type_list.append(postgres_type)
        if len(ddl_stmt) == 0:
            ddl_stmt = f"CREATE TABLE {name} (\n    "
        else:
            ddl_stmt += "\n  , "
        ddl_stmt += field["name"] + ' '
        if type(postgres_type) is str:
            ddl_stmt += postgres_type
        elif "record" in postgres_type:
            ddl_stmt += postgres_type["record"]
        else:
            ddl_stmt += postgres_type["array"]
        if not nullable:
            ddl_stmt += " NOT NULL"
        if "default" in field:
            ddl_stmt += f" DEFAULT {default_str}"
    if 0 < len(ddl_stmt):
        ddl_stmt += "\n);\n"
        ddl_queue.append(ddl_stmt)
    return postgres_type_list


def convert_to_postgres_value(avro_type, postgres_type, avro_value, null_if_convert_error):
    if avro_value is None:
        return "NULL"

    try:
        value_str = ""

        if type(avro_type) is list:
            if "null" in avro_type:
                avro_type.remove("null")
            assert len(avro_type) == 1
            avro_type = avro_type[0]

        if type(avro_type) is str:
            if avro_type == "string":
                value_str = escape(avro_value)
            elif avro_type == "long":
                value_str = str(avro_value)
            elif avro_type == "double":
                value_str = str(avro_value)
            elif avro_type == "boolean":
                if avro_value:
                    value_str = "TRUE"
                else:
                    value_str = "FALSE"
        elif type(avro_type) is dict:
            if "sqlType" in avro_type:
                sql_type = avro_type["sqlType"]
                if sql_type == "JSON":
                    value_str = escape(avro_value)
            elif "logicalType" in avro_type:
                logical_type = avro_type["logicalType"]
                if logical_type == "decimal":
                    value_str = str(avro_value)
                elif logical_type in ("timestamp-millis", "timestamp-micros"):
                    value_str = escape(avro_value.isoformat())
                elif logical_type == "date":
                    value_str = escape(avro_value.isoformat())
                elif logical_type in ("time-millis", "time-micros"):
                    value_str = escape(avro_value.isoformat())
                elif logical_type in ("local-timestamp-millis", "local-timestamp-micros"):
                    value_str = escape(avro_value.isoformat())
            else:
                avro_child_type = avro_type["type"]
                if avro_child_type == "array":
                    avro_items_type = avro_type["items"]
                    postgres_items_type = postgres_type["items"]
                    postgres_value_list = []
                    for val in avro_value:
                        postgres_value_list.append(convert_to_postgres_value(
                            avro_items_type, postgres_items_type, val, null_if_convert_error))
                    value_str = f'ARRAY[{",".join(postgres_value_list)}]::{postgres_type["array"]}'
                elif avro_child_type == "record":
                    avro_fields = avro_type["fields"]
                    postgres_fields_type = postgres_type["fields"]
                    postgres_value_list = []
                    for avro_field, postgres_field_type in zip(avro_fields, postgres_fields_type):
                        postgres_value_list.append(convert_to_postgres_value(
                            avro_field["type"], postgres_field_type, avro_value[avro_field["name"]], null_if_convert_error))
                    value_str = f'ROW({",".join(postgres_value_list)})'

        if len(value_str) == 0:
            raise MyException(f"Convert Error: type={avro_type}, value={avro_value}")
    except Exception as e:
        if null_if_convert_error:
            return "NULL"
        else:
            raise e
    else:
        return value_str


def make_sql_insert(tablename, schema, postgres_type_list, rec, null_if_convert_error=False):
    insert_into = []
    insert_values = []
    fields = schema["fields"]
    for field, postgres_type in zip(fields, postgres_type_list):
        field_name = field["name"]
        insert_into.append(field_name)
        insert_values.append(convert_to_postgres_value(field["type"], postgres_type, rec[field_name], null_if_convert_error))
    insert_into_str = "\n  , ".join(insert_into)
    insert_values_str = "\n  , ".join(insert_values)
    insert_stmt = f"INSERT INTO {tablename} (\n    {insert_into_str}\n)\nVALUES (\n    {insert_values_str}\n);\n"
    return insert_stmt


def main():
    in_path = Path(sys.argv[1])
    out_dir = Path.cwd()    tablename = in_path.stem.lower()
    postgres_record_type_prefix = tablename
    out_path = out_dir / f"{tablename}.sql"

    with open(in_path, "rb") as fi, open(out_path, "wt", encoding="utf-8") as fo:
        reader = MyReader(fi)
        schema = json.loads(reader.meta["avro.schema"].decode("utf-8"))

        # トランザクション開始
        fo.write("BEGIN;\n")

        # DDL
        ddl_queue = deque()
        postgres_type_list = make_sql_create_table(tablename, schema["fields"], ddl_queue, postgres_record_type_prefix)
        for ddl_stmt in ddl_queue:
            fo.write(ddl_stmt)

        # INSERT文
        num = 0
        for rec in reader:
            num += 1
            insert_stmt = make_sql_insert(tablename, schema, postgres_type_list, rec, null_if_convert_error=False)
            fo.write(insert_stmt)
        print(f"{num} 行")

        # トランザクション終了
        fo.write("COMMIT;\n")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        print(traceback.format_exc())

このPythonコードの起動例です。コマンドライン引数で渡せるAvroファイルは1つのみです。


python bq_avro_to_postgres_db.py events_20210131.avro

以下のように、DDLの後にINSERT文が入ったSQLファイルが出力されます。

events_20210131.sql


BEGIN;
CREATE TYPE events_20210131_type_2 AS (
    string_value VARCHAR
  , int_value BIGINT
  , float_value DOUBLE PRECISION
  , double_value DOUBLE PRECISION
);
CREATE TYPE events_20210131_type_1 AS (
    key VARCHAR
  , value events_20210131_type_2
);
CREATE TYPE events_20210131_type_3 AS (
    analytics_storage BIGINT
  , ads_storage BIGINT
  , uses_transient_token VARCHAR
);
CREATE TYPE events_20210131_type_5 AS (
    string_value BIGINT
  , int_value BIGINT
  , float_value BIGINT
  , double_value BIGINT
  , set_timestamp_micros BIGINT
);
CREATE TYPE events_20210131_type_4 AS (
    key BIGINT
  , value events_20210131_type_5
);
CREATE TYPE events_20210131_type_6 AS (
    revenue DOUBLE PRECISION
  , currency VARCHAR
);
CREATE TYPE events_20210131_type_8 AS (
    browser VARCHAR
  , browser_version VARCHAR
);
CREATE TYPE events_20210131_type_7 AS (
    category VARCHAR
  , mobile_brand_name VARCHAR
  , mobile_model_name VARCHAR
  , mobile_marketing_name VARCHAR
  , mobile_os_hardware_model BIGINT
  , operating_system VARCHAR
  , operating_system_version VARCHAR
  , vendor_id BIGINT
  , advertising_id BIGINT
  , language VARCHAR
  , is_limited_ad_tracking VARCHAR
  , time_zone_offset_seconds BIGINT
  , web_info events_20210131_type_8
);
CREATE TYPE events_20210131_type_9 AS (
    continent VARCHAR
  , sub_continent VARCHAR
  , country VARCHAR
  , region VARCHAR
  , city VARCHAR
  , metro VARCHAR
);
CREATE TYPE events_20210131_type_10 AS (
    id VARCHAR
  , version VARCHAR
  , install_store VARCHAR
  , firebase_app_id VARCHAR
  , install_source VARCHAR
);
CREATE TYPE events_20210131_type_11 AS (
    medium VARCHAR
  , name VARCHAR
  , source VARCHAR
);
CREATE TYPE events_20210131_type_12 AS (
    hostname VARCHAR
);
CREATE TYPE events_20210131_type_13 AS (
    total_item_quantity BIGINT
  , purchase_revenue_in_usd DOUBLE PRECISION
  , purchase_revenue DOUBLE PRECISION
  , refund_value_in_usd DOUBLE PRECISION
  , refund_value DOUBLE PRECISION
  , shipping_value_in_usd DOUBLE PRECISION
  , shipping_value DOUBLE PRECISION
  , tax_value_in_usd DOUBLE PRECISION
  , tax_value DOUBLE PRECISION
  , unique_items BIGINT
  , transaction_id VARCHAR
);
CREATE TYPE events_20210131_type_14 AS (
    item_id VARCHAR
  , item_name VARCHAR
  , item_brand VARCHAR
  , item_variant VARCHAR
  , item_category VARCHAR
  , item_category2 VARCHAR
  , item_category3 VARCHAR
  , item_category4 VARCHAR
  , item_category5 VARCHAR
  , price_in_usd DOUBLE PRECISION
  , price DOUBLE PRECISION
  , quantity BIGINT
  , item_revenue_in_usd DOUBLE PRECISION
  , item_revenue DOUBLE PRECISION
  , item_refund_in_usd DOUBLE PRECISION
  , item_refund DOUBLE PRECISION
  , coupon VARCHAR
  , affiliation VARCHAR
  , location_id VARCHAR
  , item_list_id VARCHAR
  , item_list_name VARCHAR
  , item_list_index VARCHAR
  , promotion_id VARCHAR
  , promotion_name VARCHAR
  , creative_name VARCHAR
  , creative_slot VARCHAR
);
CREATE TABLE events_20210131 (
    event_date VARCHAR DEFAULT NULL
  , event_timestamp BIGINT DEFAULT NULL
  , event_name VARCHAR DEFAULT NULL
  , event_params events_20210131_type_1[] NOT NULL
  , event_previous_timestamp BIGINT DEFAULT NULL
  , event_value_in_usd DOUBLE PRECISION DEFAULT NULL
  , event_bundle_sequence_id BIGINT DEFAULT NULL
  , event_server_timestamp_offset BIGINT DEFAULT NULL
  , user_id VARCHAR DEFAULT NULL
  , user_pseudo_id VARCHAR DEFAULT NULL
  , privacy_info events_20210131_type_3 DEFAULT NULL
  , user_properties events_20210131_type_4[] NOT NULL
  , user_first_touch_timestamp BIGINT DEFAULT NULL
  , user_ltv events_20210131_type_6 DEFAULT NULL
  , device events_20210131_type_7 DEFAULT NULL
  , geo events_20210131_type_9 DEFAULT NULL
  , app_info events_20210131_type_10 DEFAULT NULL
  , traffic_source events_20210131_type_11 DEFAULT NULL
  , stream_id BIGINT DEFAULT NULL
  , platform VARCHAR DEFAULT NULL
  , event_dimensions events_20210131_type_12 DEFAULT NULL
  , ecommerce events_20210131_type_13 DEFAULT NULL
  , items events_20210131_type_14[] NOT NULL
);
INSERT INTO events_20210131 (
    event_date
  , event_timestamp
  , event_name
  , event_params
  , event_previous_timestamp
  , event_value_in_usd
  , event_bundle_sequence_id
  , event_server_timestamp_offset
  , user_id
  , user_pseudo_id
  , privacy_info
  , user_properties
  , user_first_touch_timestamp
  , user_ltv
  , device
  , geo
  , app_info
  , traffic_source
  , stream_id
  , platform
  , event_dimensions
  , ecommerce
  , items
)
VALUES (
    '20210131'
  , 1612069510766593
  , 'page_view'
  , ARRAY[ROW('gclid',ROW(NULL,NULL,NULL,NULL)),ROW('gclsrc',ROW(NULL,NULL,NULL,NULL)),ROW('debug_mode',ROW(NULL,1,NULL,NULL)),ROW('ga_session_number',ROW(NULL,1,NULL,NULL)),ROW('all_data',ROW(NULL,NULL,NULL,NULL)),ROW('page_location',ROW('https://shop.googlemerchandisestore.com/',NULL,NULL,NULL)),ROW('entrances',ROW(NULL,1,NULL,NULL)),ROW('session_engaged',ROW('0',NULL,NULL,NULL)),ROW('ga_session_id',ROW(NULL,661084800,NULL,NULL)),ROW('clean_event',ROW('gtm.js',NULL,NULL,NULL)),ROW('engaged_session_event',ROW(NULL,1,NULL,NULL)),ROW('page_title',ROW('Home',NULL,NULL,NULL))]::events_20210131_type_1[]
  , NULL
  , NULL
  , 6595101026
  , NULL
  , NULL
  , '1026454.4271112504'
  , ROW(NULL,NULL,'No')
  , ARRAY[]::events_20210131_type_4[]
  , 1612069510766593
  , ROW(0.0,'USD')
  , ROW('mobile','Apple','iPhone','<Other>',NULL,'Web','<Other>',NULL,NULL,'en-us','No',NULL,ROW('Safari','13.1'))
  , ROW('Americas','Northern America','United States','California','San Carlos','(not set)')
  , NULL
  , ROW('organic','(organic)','google')
  , 2100450278
  , 'WEB'
  , NULL
  , ROW(NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL)
  , ARRAY[]::events_20210131_type_14[]
);

（中略）

COMMIT;

生成されたSQL文をPostgreSQLで動かしてimportする

SQL文の内容を確認し、必要に応じてSQL文や上記のPythonコードを変更すると良いでしょう。
その後、SQL文をPostgreSQLで実行してimportします。

以下の画像は、importされたテーブルをDataGripで確認しているところです。

BigQueryとPostgreSQLで同じSELECT文を実行して、同じ結果が返ることをざっくり確認する

本記事でPostgreSQLにimportしたのは、GA4のeコマースサンプルデータセットです。
UNNEST関数を使ったクエリーを2つ試してみます。
FROM句のテーブル名だけ、BigQuery用とPostgreSQL用で修正する必要があります。

クエリーその1


WITH
    foo AS (
        SELECT
            event_date,
            event_timestamp,
            event_name,
            event_params
        FROM
            public.events_20210131
        WHERE
            event_timestamp IN (1612125113370459, 1612125041889438)
    )
SELECT
    event_date,
    event_timestamp,
    event_name,
    event_param.key,
    (event_param.value).string_value,
    (event_param.value).int_value,
    (event_param.value).float_value,
    (event_param.value).double_value
FROM
    foo
CROSS JOIN
    UNNEST(event_params) AS event_param
ORDER BY
    event_timestamp,
    event_name,
    key

BigQueryの結果：

PostgreSQLの結果：

両方とも21件、同じデータが返りました。

クエリーその2


SELECT
    event_date,
    event_timestamp,
    event_name,
    (SELECT
            (value).int_value
        FROM
            UNNEST(event_params)
        WHERE
            key = 'ga_session_id'
        ) AS ga_session_id,
    (SELECT
            (value).string_value
        FROM
            UNNEST(event_params)
        WHERE
            key = 'page_title'
        ) AS page_title
FROM
    public.events_20210131
WHERE
    event_timestamp IN (1612125113370459, 1612125041889438)
ORDER BY
    event_timestamp,
    event_name

BigQueryの結果：

PostgreSQLの結果：

両方とも2件、同じデータが返りました。

2つのクエリーで、BigQueryとPostgreSQLが同じ結果を返すことを、ざっくり確認できました。

続編（予告）

本記事のPythonコードは、GA4のテーブルコピー専用ではなく、汎用を目指したものです。
次回はGA4のテーブル専用のコードを書いて、PostgreSQL側でレンジパーティションを使用することを考えています。

追記（2022-12-31）

予告した記事を投稿しました。
GA4のテーブルをBigQueryからPostgreSQLにAvroファイル経由でコピーする