FileIOのDynamic DestinationのDestinationTはequals/hashCodeが必要だよ

細かすぎて伝わらないApache Beam/Dataflow選手権#1です。

# FileIOって？
Apache Beamのファイル一般を扱うIOクラスです。
[ドキュメント](https://beam.apache.org/documentation/io/built-in/)によるとS3/GCS/HDFS/ローカルのファイルを、扱う事が出来るようです。

# Dynamic Destinationって？
PCollectionの中身によって出力先を変える機能です。
具体的な設定は[FileIO.Write](https://beam.apache.org/releases/javadoc/2.15.0/org/apache/beam/sdk/io/FileIO.Write.html)で定義されており、


* [by](https://beam.apache.org/releases/javadoc/2.15.0/org/apache/beam/sdk/io/FileIO.Write.html#by-org.apache.beam.sdk.transforms.Contextful-)で、入力(UserT)をグループ(DestinationT)に分類するクラスを指定
* [via](https://beam.apache.org/releases/javadoc/2.15.0/org/apache/beam/sdk/io/FileIO.Write.html#via-org.apache.beam.sdk.transforms.Contextful-org.apache.beam.sdk.io.FileIO.Sink-a)で、入力(UserT)を出力(OutputT)に変換するクラスを指定
* [withNaming](https://beam.apache.org/releases/javadoc/2.15.0/org/apache/beam/sdk/io/FileIO.Write.html#withNaming-org.apache.beam.sdk.transforms.SerializableFunction-)で、グループ(DestinationT)からファイル名を決めるクラスを指定

の３つが、Dynamic Destinationの主な設定です。

# DestinationTって
入力(UserT)分類し、グループ化するためのラベルとして使われるクラス（ジェネリクス）です。
境界(extends)が定義されてはいないので、特に必要なメソッドや親クラスは指定されていません。

**って思うじゃん‥**

# 詰まった
# やろうとしていたこと
* 数十万件のPub/Subメッセージを、Dynamic Destinationで仕分け（5種類くらい）してGCSに書き込みたい
* Dynamic Destinationなしでは、楽勝に処理出来ていた

## 起きたこと
* ファイルを書き出す段階で、WorkerでOut Of Memory(OOM)が起きた
 * （Dynamic Destination無しと、同じインスタンス種類の場合）
* ごく一部しか結果がファイルに書き込まれなかった
* ワーカーのメモリを巨大（64GB）にすると、ゆっくり書き出される
  * オートスケールにしても台数は増えない

## 原因
DestinationTの実装に

* equals
* hashCode

を（正しく）実装していなかったのが原因。

## なぜ必要か
端的にはDestinationTがHashMap[で使われている](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L754)ためです。

### 細かい流れが気になる人向けに：
Dynamic Destinationでは、以下の流れで書き込みを行います。

1. 入力(UserT)をファイルに割り振るために、番号を付ける（[ここらへん](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L667)）
 * シャードの連番とDestinationTをencodeした結果のハッシュ[を使っています](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L707)
 * encodeには[Coderのencodeメソッド](https://qiita.com/kurtosis/items/23d3c46deb6bf4455d18)が使われます
1. 番号でGroupByKey([ここらへん](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L670))
1. 一時ファイルに書き出す（[ここらへん](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L746)）
1. 一時ファイルを改名（[ここらへん](https://github.com/apache/beam/blob/7931ec055e2da7214c82e368ef7d7fd679faaef1/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L829)）

下のような理由で、DestinationTがHashMapに使われています：

* 入力を連番に割り振る時(1)に、ハッシュが衝突する可能性がある
* 衝突すると、異なるDestinationがGroupByKeyでまとめられる
* そのため、一時ファイル書き出し(3)の部分でもDestinationTを計算。異なるDestinationTを異なるファイルに書き出すようにしている
 * この時、DestinationTと書き出し先を管理するために、DestinationTをHashMapに突っ込んでいる

このため、DestinationTがhashCode/equalsを（正しく）実装していないと、

* 入力毎に違う書き出し先とみなされる
 * つまり1ファイル1行
* 超絶細かい単位で書き出すので時間がかかる
* 出力が終わるまで、入力データが保持するので、メモリ使用量が多くなる

のではないかと推測しています。

# 今回学んだこと

* equals/hashCodeを実装しないでHashMapを使うと死ぬ
 * Effective Java大事
* DataflowでもDumpが取れる
 * Googleの人が[わかりやすい資料](https://github.com/GoogleCloudPlatform/community/blob/f9d6dc8ec47f78f3a18a0c0ac82acd66bfb1c82e/tutorials/dataflow-debug-oom-conditions/index.md)を書いてくれています
 * （がダンプしても分かるとは限らない）
* Beamのファイル書き出しは、ちょっと複雑な処理をしている