CSVファイルで消耗しないためのApache Avro入門

# はじめに
こんにちは、[株式会社unerry](https://www.unerry.co.jp/)でデータエンジニアをしているJanosと申します。
弊社ではIoTセンサーのBeaconによる実社会の位置情報を中心としたサービスを展開しています。私はデータエンジニアとして普段他社とのデータ連携処理を実装することが多く、例えば今回のようにcsvファイルを受領しBigQueryへloadするようなタスクは往々にして発生します。この時連携するデータは多くのケースで非常に大きいサイズとなっています。

## issue
今回の問題についてまとめると、まず31日間のサンプルデータを```csv.gz```で受領したのですが

- １ファイルあたり約5~7MiB
- １日あたり200ファイル

というような構造になっており１日あたり約10~11Gib、総計１ヶ月全体で約300 GiBというデータ量でした。

これら全てを分析のためBigQueryへloadしていきます。PythonでcsvをBigQueryにloadするにあたりコードをそんなに書く必要はありません。

```python
job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("name", "STRING"),
        bigquery.SchemaField("post_abbr", "STRING"),
    ],
    skip_leading_rows=1,
    source_format=bigquery.SourceFormat.CSV,
)

uri = "gs://cloud-samples-data/bigquery/us-states/us-states.csv"

load_job = client.load_table_from_uri(
    uri, table_id, job_config=job_config
) 
```

ところが```load_table_from_uri```はcsvのparseに失敗したりcsvのrowにschemaで指定された型と違う値が入っていたりすると全体が失敗してしまいます。失敗すると今回の処理のどこで失敗したか教えてくれるのですが、この先ファイルのどこで失敗するのかは教えてくれません。毎回１つのエラーだけです。（参考：https://googleapis.dev/python/bigquery/latest/generated/google.cloud.bigquery.client.Client.html#google.cloud.bigquery.client.Client.load_table_from_uri）

こうなるとload処理は大変なものになってしまいます。全体でどれくらい不正なrowがあるかも分からず、また不正なrowがあったとしてその原因を突き止めて修正なり除去なりをしようとしても一つのファイルが巨大すぎて効率よく対処できません。

回避する方法として```max_bad_records```として無視する不正なrowの最大数を指定するというものがあります。しかし巨大なファイル群を前にして不正なrow数が予めわかっていることは殆どなく、これでは神頼みのようになってしまいます。

# Apache Avro形式とは
今回この問題をcsvから[Apache Avro](https://avro.apache.org/)形式に変換することで解決しました。Apache Avroはバイナリ形式にシリアライズしたデータとそのデータのスキーマを同一ファイル内にバンドルするファイル形式です。

次はAvroで用いられるスキーマの例を示します。スキーマはJSONで定義することができます：

```json
{"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}    
```

Avro形式を利用することの最大の利点はこの型を指定できるスキーマにありました。csvと違い明確に型を指定することで、BigQueryにloadする際不正なデータが混入することを防ぐことが可能になったのです。

## BigQueryとAvro
またAvro形式はBigQueryへデータをloadすること自体に適しています。データを並列に読み取ることができる為、高速にBigQueryへloadすることができます。

https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-avro#advantages_of_avro

# CSVからAvroへの変換
ではいよいよ実際にcsvからAvroへ変換する方法を見ていきます。

まずはAvroスキーマの定義からです。

```python
fields = [
    {"name": "id",
     "type": "string"},
    {"name": "name",
     "type": ["null", "string"],
     "default": None},
    {"name": "weight",
     "type": ["null", "float"],
     "default": None},
    {"name": "height",
     "type": ["null", "float"],
     "default": None},
]

schema = {
    "doc": "this is sample data",
    "name": "Data",
    "namespace": "example",
    "type": "record",
    "fields": fields,
}

```
null値を許容するcolumnは```["null", "string"]```のようにそのcolumnの型とnullのリストを指定します。

まずはcsvを読み込みたいのですが、これには[pandas](https://pandas.pydata.org/)を用いることにします。

```python
import pandas as pd

columns = ["id", "name", "weight", "height"]

df = pd.read_csv(buffer, error_bad_lines=False)
df.columns = columns
records = df.to_dict("records")
```
Avroに書き込むときに``` [{column -> value_0}, … , {column -> value_n}]```の形のdictのlistが必要になるので```df.to_dict("records")```でこれを生成しています。

今このrecordsの中には不正なrowが混入している状態です。今回Avro形式を採用した恩恵はschemaに沿ったvalidationを実行することにより浴することができます。validate関数を作りfilterすることでschemaに適合しない不正なrowを除去します。

```python 
import fastavro

def validate(x) -> bool:
    return fastavro.validate(x, schema, raise_errors=False)

good_records = filter(validate, records)
```

Avroの書き込みにはより高速と言われている[fastavro](https://fastavro.readthedocs.io/en/latest/)を用いています。今回は変換したAvroファイルをGCSへアップロードしました。

```python
import fastavro
from google.cloud import storage


buffer = io.BytesIO()
fastavro.writer(buffer, schema=schema,
                records=good_records)
buffer.seek(0)

storage_client = storage.Client()
bucket = storage_client.bucket(bucketname)
blob = bucket.blob(target_uri)
try:
    blob.upload_from_string(buffer.read())
except Exception as e:
    logging.error(e)
```

以上でcsvからAvroへの変換が出来ました。実際変換したAvroファイルをBigQueryにまとめてloadするとデータ一日あたり数分で終わり、途中でエラーで躓くこともなく無事に読み込むことができました。

# 日付はどう扱う？
最後にAvroで日付をどう扱うかについて触れておこうと思います。日付を表すDateはAvroではlogical typeというprimitiveまたはcomplexの型に付加的な属性を持つもののひとつとして定義されます。

```json
{"name": "measuring_date",
 "type": {
     "type": "int",
     "logicalType": "date"
 }}
```
AvroではDateの値は1970年1月1日からの経過日数のintとして表現されます。日付を変換する際には一手間必要になるかと思います。