BigQueryに読ませるAvroファイルをfastavroで作るときの、Pythonでのスキーマの書き方

Google BigQueryにデータをimport/exportするときに、私はApache Avro形式のファイルを使っています。
Avro形式は、データだけでなくテーブルのスキーマ情報も持つので、テーブル作成のデータソースとしてjsonやcsvより便利です。
また、バイナリファイルなのでjsonやcsvよりも高速に処理できるそうです。

BigQueryへのimport用にAvroファイルを作成するとき、私は[fastavro](https://fastavro.readthedocs.io/en/latest/)を使用して以下のようなPythonコードを書いています。
下記コードでは変数名SCHEMA内のfields配列に各列の情報を入れているのですが、本記事では BigQueryのそれぞれの型に対応するPythonコードの書き方を、わかる範囲でまとめてみました。
（記述に誤りがありましたらご指摘くださいませ）

```py

import traceback
import fastavro
from pathlib import Path

SCHEMA = {
    "type": "record",
    "name": "Root",
    "fields": [
        {
            "name": "id",
            "type": "long",
        },
        {
            "name": "rating",
            "type": "long",
        },
        {
            "name": "title",
            "type": "string",
        },
        {
            "name": "comment",
            "type": "string",
        }
    ]
}


DATA = [
    {
        "id": 1,
        "rating": 5,
        "title": "最高です",
        "comment": "感動しました"
    },
    {
        "id": 2,
        "rating": 4,
        "title": "良いです",
        "comment": "気に入りました"
    }
]


def main():

    out_dir = Path.cwd()
    out_path = out_dir / "foo.avro"

    with open(out_path, "wb") as fo:
        writer = fastavro.write.Writer(fo, SCHEMA)
        for rec in DATA:
            writer.write(rec)
        writer.flush()


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

それでは型マッピングの一覧です。
BIGNUMERIC型とGEOGRAPHY型は、試した結果よくわからない結果になったので、割愛しました。
なお、全体を通じて "name" に設定する値は任意です。

# プリミティブ型

## モード=REQUIRED

### タイプ=STRING

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/74c7c9be-f120-e292-3004-cec690c37ad9.png)

- Python

```py

        {
            "name": "col_string",
            "type": "string"
        },
```

### タイプ=BYTES

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/71c9188f-7171-74e0-8fdc-2e18fbd39513.png)

- Python

```py
        {
            "name": "col_bytes",
            "type": "bytes"
        },
```

### タイプ=INTEGER

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/ed711591-4e24-3f27-538c-28df1f548be8.png)

- Python

```py

        {
            "name": "col_integer",
            "type": "long"
        },
```

### タイプ=FLOAT

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/d644679f-5364-307c-0d40-3165ed03c75f.png)

- Python

```py

        {
            "name": "col_float",
            "type": "double"
        },
```

### タイプ=NUMERIC

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/d78f9fc6-e4d9-1f39-925b-47133117928b.png)

- Python

```py

        {
            "name": "col_numeric",
            "type": {
                "type": "bytes",
                "logicalType": "decimal",
                "precision": 38,
                "scale": 9
            }
        },
```

### タイプ=BOOLEAN

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/d3347b46-81da-63ac-ea67-23b3b90d6303.png)

- Python

```py

        {
            "name": "col_boolean",
            "type": "boolean"
        },
```

### タイプ=TIMESTAMP

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/78369f8a-42d0-7e47-d360-4dc219f5e27a.png)

- Python

```py
        {
            "name": "col_timestamp",
            "type": {
                "type": "long",
                "logicalType": "timestamp-millis"
            }
        }
```

```py

        {
            "name": "col_timestamp",
            "type": {
                "type": "long",
                "logicalType": "timestamp-micros"
            }
        },
```

### タイプ=DATE

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/eb8d8634-bd95-6ca6-5b27-3481400dcb2e.png)

- Python

```py

        {
            "name": "col_date",
            "type": {
                "type": "int",
                "logicalType": "date"
            }
        },
```

ただし、DATE型をBigQueryからavroファイルに出力するとstring型になる。

### タイプ=TIME

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/55a96c79-bf0d-a9a3-894c-eb0619f8a9e3.png)

- Python

```py
        {
            "name": "col_time",
            "type": {
                "type": "long",
                "logicalType": "time-micros"
            }
        },
```

ただし、TIME型をBigQueryからavroファイルに出力するとstring型になる。

### タイプ=DATETIME

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/8456fe8b-e677-15bf-76f5-8804c4677a1c.png)

- Python

```py

        {
            "name": "col_datetime",
            "type": {
                "type": "long",
                "logicalType": "local-timestamp-millis"
            }
        },
```

```py

        {
            "name": "col_datetime",
            "type": {
                "type": "long",
                "logicalType": "local-timestamp-micros"
            }
        },
```

ただし、DATETIME型をBigQueryからavroファイルに出力するとstring型になる。

### タイプ=JSON
- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/117e88c6-fd4f-25c9-b1c2-982f59d0c0c0.png)

- Python

```py
        {
            "name": "col_json",
            "type": {
                "type": "string",
                "sqlType": "JSON"
            }
        },
```

## モード=REQUIRED以外

### モード=NULLABLE

- BigQuery![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/18ce3f41-9bfa-2537-45d1-0f17be6a2685.png)

- Python

```py

        {
            "name": "col_string_nullable",
            "type": [
                "null",
                "string"
            ],
            "default": null
        },
```

### モード=REPEATED

- BigQuery![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/cc671eb7-5cb3-e691-4ea2-a9b1ee4a02c2.png)

- Python

```py

        {
            "name": "col_string_repeated",
            "type": {
                "type": "array",
                "items": "string"
            }
        },
```

# レコード型

## タイプ=RECORD、モード=REQUIRED

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/ae41d8f1-e513-cc81-96cd-f3751a01e67b.png)

- Python

```py

        {
            "name": "col_record",
            "type": {
                "type": "record",
                "namespace": "root",
                "name": "Col_Record",
                "fields": [
                    {
                        "name": "col_record_1",
                        "type": "string"
                    },
                    {
                        "name": "col_record_2",
                        "type": "string"
                    }
                ]
            }
        },
```

## タイプ=RECORD、モード=NULLABLE

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/a0ee96a8-f699-82b9-bc7e-7722f1ebd13d.png)

- Python

```py

        {
            "name": "col_record_nullable",
            "type": [
                "null",
                {
                    "type": "record",
                    "namespace": "root",
                    "name": "Col_Record_Nullable",
                    "fields": [
                        {
                            "name": "col_record_1",
                            "type": "string"
                        },
                        {
                            "name": "col_record_2",
                            "type": "string"
                        }
                    ]
                }
            ],
            "default": null
        }
```

## タイプ=RECORD、モード=REPEATED

- BigQuery
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/2dbe9a3f-75cf-abdd-9c8d-1a007f3d98ae.png)

- Python

```py

        {
            "name": "col_record_repeated",
            "type": {
                "type": "array",
                "items": {
                    "type": "record",
                    "namespace": "root",
                    "name": "Col_Record_Repeated",
                    "fields": [
                        {
                            "name": "col_record_1",
                            "type": "string"
                        },
                        {
                            "name": "col_record_2",
                            "type": "string"
                        }
                    ]
                }
            }
        },
```

以上の型マッピングの情報を用いて、実際にPythonコードを書いてみます。

# Avroファイルを作って BigQuery に import してみる

以下のPythonコードを実行して Avroファイルを作ります。

```py

import traceback
import fastavro
from pathlib import Path
import time
import datetime
from decimal import *


SCHEMA = {
    "type": "record",
    "name": "Root",
    "fields": [
        {
            "name": "col_string",
            "type": "string"
        },
        {
            "name": "col_bytes",
            "type": "bytes"
        },
        {
            "name": "col_integer",
            "type": "long"
        },
        {
            "name": "col_float",
            "type": "double"
        },
        {
            "name": "col_numeric",
            "type": {
                "type": "bytes",
                "logicalType": "decimal",
                "precision": 38,
                "scale": 9
            }
        },
        {
            "name": "col_boolean",
            "type": "boolean"
        },
        {
            "name": "col_timestamp_millis",
            "type": {
                "type": "long",
                "logicalType": "timestamp-millis"
            }
        },
        {
            "name": "col_timestamp_micros",
            "type": {
                "type": "long",
                "logicalType": "timestamp-micros"
            }
        },
        {
            "name": "col_date",
            "type": {
                "type": "int",
                "logicalType": "date"
            }
        },
        {
            "name": "col_time",
            "type": {
                "type": "long",
                "logicalType": "time-micros"
            }
        },
        {
            "name": "col_datetime_millis",
            "type": {
                "type": "long",
                "logicalType": "local-timestamp-millis"
            }
        },
        {
            "name": "col_datetime_micros",
            "type": {
                "type": "long",
                "logicalType": "local-timestamp-micros"
            }
        },
        {
            "name": "col_json",
            "type": {
                "type": "string",
                "sqlType": "JSON"
            }
        },
        {
            "name": "col_string_nullable",
            "type": [
                "null",
                "string"
            ],
            "default": None
        },
        {
            "name": "col_string_repeated",
            "type": {
                "type": "array",
                "items": "string"
            }
        },
        {
            "name": "col_record",
            "type": {
                "type": "record",
                "namespace": "root",
                "name": "Col_Record",
                "fields": [
                    {
                        "name": "col_record_1",
                        "type": "string"
                    },
                    {
                        "name": "col_record_2",
                        "type": "string"
                    }
                ]
            }
        },
        {
            "name": "col_record_nullable",
            "type": [
                "null",
                {
                    "type": "record",
                    "namespace": "root",
                    "name": "Col_Record_Nullable",
                    "fields": [
                        {
                            "name": "col_record_1",
                            "type": "string"
                        },
                        {
                            "name": "col_record_2",
                            "type": "string"
                        }
                    ]
                }
            ],
            "default": None
        },
        {
            "name": "col_record_repeated",
            "type": {
                "type": "array",
                "items": {
                    "type": "record",
                    "namespace": "root",
                    "name": "Col_Record_Repeated",
                    "fields": [
                        {
                            "name": "col_record_1",
                            "type": "string"
                        },
                        {
                            "name": "col_record_2",
                            "type": "string"
                        }
                    ]
                }
            }
        }
    ]
}


def main():

    out_dir = Path.cwd()
    out_path = out_dir / "foo.avro"

    with open(out_path, "wb") as fo:
        writer = fastavro.write.Writer(fo, SCHEMA)
        writer.write({
            "col_string"           : "string",
            "col_bytes"            : b"abc",
            "col_integer"          : -1234567890,
            "col_float"            : 3.14,
            "col_numeric"          : Decimal("123.456"),
            "col_boolean"          : True,
            "col_timestamp_millis" : datetime.datetime(2022, 1, 2, 3, 4, 5, 6),
            "col_timestamp_micros" : datetime.datetime(2022, 1, 2, 3, 4, 5, 6),
            "col_date"             : datetime.date(2022, 1, 2),
            "col_time"             : time.time() * 1_000_000,
            "col_datetime_millis"  : datetime.datetime(2022, 1, 2, 3, 4, 5, 6),
            "col_datetime_micros"  : datetime.datetime(2022, 1, 2, 3, 4, 5, 6),
            "col_json"             : '{"key": 1, "value": "あ"}',
            "col_string_nullable"  : None,
            "col_string_repeated"  : ["a", "b", "c"],
            "col_record"           : {"col_record_1":"A", "col_record_2":"B"},
            "col_record_nullable"  : None,
            "col_record_repeated"  : [{"col_record_1":"A", "col_record_2":"B"},{"col_record_1":"C", "col_record_2":"D"}]
        })
        writer.flush()


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

実行すると foo.avro が生成されました。
以下のコマンドを使って BigQuery にロードします。

```

bq load \
    --replace \
    --source_format=AVRO \
    --use_avro_logical_types \
    --project_id=example_project \
    example_dataset.foo \
    ./foo.avro
```

以下のように、意図したスキーマで fooテーブルが作成されました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/926daa83-9bab-9d6d-5a6a-0a8cb360bfe4.png)

fooテーブルのデータを見てみます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/52f982f8-70d2-0ca4-b3e9-03374801e94b.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/327312/ecae146c-2032-e043-816a-26f6bf440d66.png)

Pythonコードで設定したデータがロードされているようです。

# 参考URL
* [BigQueryやPythonからAvroファイルに出力しようとしてDATETIME型でつまづいた](https://qiita.com/kanedaq/items/ce8c8a7b959ea739a3f8)