Spark Datasetの再帰構造を扱うのに苦労した件

今日もみんな大好きimplicitのお話ですよ（出番は少なめ）。

# 1. 何がしたかった？
[前の記事](https://qiita.com/ayatty/items/660157cdfb3d77cf7fc5)でも書いた通り、Datasetに対してグラフ的な繰り返し処理をしたかった。
まず起点となるレコードを選択して、そのレコードに隣接するレコードが**条件**を満たす場合は隣接レコードに進み・・・を繰り返して、目的のレコードに到達したい。

その**条件**というのが1ホップ毎に変わっていくので、

1. 0ホップ目のレコード + 条件1 → 1ホップ目のレコード
2. 1ホップ目のレコード + 条件2 → 2ホップ目のレコード
2. 2ホップ目のレコード + 条件3 → 3ホップ目のレコード
2. 3ホップ目のレコード + 条件4 → 4ホップ目のレコード
（以下略）

こんな感じの繰り返しになるので、条件を最初はリストで

```
[条件1, 条件2, 条件3, 条件4,,,,]
```
のような形で持たせていましたが、やってるうちにもっと複雑になってきてしまって、

```
[条件1, 条件2, 条件3, 条件4,,,,]
[条件1, 条件2, 条件A]
```
みたい感じになってきた。
2回ループを回すのももったいないし、条件1と条件2までは同じ判定をすればいいので、1回で済むところは済ませたい。
ということで、以下のような条件を定義してDataset化して渡そうと考えたわけです。

```
(条件1, next=[
  (条件2, next=[
    (条件3, next=[条件4]),
    (条件A, next=[])
  ])
])
```
わかりますかね？条件2のnextに、条件3と条件Aが並列で入っているような形です。

具体的には以下のようなcase classを定義しました。

```scala
case class Recursive(condition: String, next: List[Recursive])
case class Message(id: Int, conditions: Recursive)
```
idは条件を適用してみるレコードのIDです。レコードは以下のようになります。

```scala
case class Record(id: Int, value: String, neighbor: List[Int])
```
id同士でjoinして、Record.valueとRecursive.conditionが一致してたら、次はneighborに対して、nextを適用する。そんな感じです。

# 2. 何がダメだったか？
いつも通りIntelliJで文句言われたところは全部つぶしてさあ実行・・・
```Exception in thread "main" java.lang.UnsupportedOperationException: cannot have circular references in class, but got the circular reference of class Recursive```
・・・オゥ。
循環参照を持ったクラスはダメなんだってさ。

# 3. 神Encoderの出番ですよ
Encoderというのは、Sparkにおけるシリアライズのための仕組みですね。詳細（でもないが）は[前回参照](https://qiita.com/ayatty/items/660157cdfb3d77cf7fc5)。上のエラーはSparkのデフォルトのEncoderがcircular referenceに対応してないよと言ってるだけで、自前で用意してあげればなんとかなる場合がある（大抵なんとかなる？）。
といっても自前で作るのは敷居が高そうなので使えるものを探してググってみて出てきたのが、神羅万象あらゆるものをエンコードするという伝説のkryo氏。

https://stackoverflow.com/questions/54241088/spark-using-recursive-case-class

こんな感じで使うそうです（implicitバージョン）。

```scala
import org.apache.spark.sql.Encoders

// Dataset化するのはRecursiveでなくMessageなのでMessageのエンコーダを定義
implicit val enc = Encoders.kryo[Message]
val msg = List(
  Message(1, Recursive("条件1", List(Recursive("条件2", List())))),
  Message(2, Recursive("条件3", List()))
).toDS()
```

これでエラーになってた大部分はつぶれました。
しかしこれで終わりではなかったのです。

# 4. ばいなりってなんですか？
問題はjoinWithに差し掛かった時でした。
[joinWith](https://spark.apache.org/docs/2.4.4/api/java/org/apache/spark/sql/Dataset.html#joinWith-org.apache.spark.sql.Dataset-org.apache.spark.sql.Column-)(公式ドキュメント)

前出の通り、RecordとMessageをidを使ってjoinWithしようとしたところ、
```idなんてカラム名は存じ上げません・・・```
みたいな感じのことを言われるわけです。

え～、エンコーダー定義したじゃなーい。
と、思ってMessageに対してprintSchemaしてみると、

```scala
msg.printSchema
// root
//  |-- value: binary (nullable = true)
```

binary！？知らないけど！？
カラムなんてなくてbinaryだけがぽつんと存在している状態。
確かにidはないけど・・・。

後で理解したことだが、これは要するに、

- 通常状態では、binaryとして格納されていて、
- 必要時(mapとかfilterとか)にエンコードされる、が、
- joinWithみたいにカラム名をStringで指定する系のヤツ（試してないけどgroupByKeyとか？）には忖度してくれない

という仕組みになっているらしい。

# 5. 解決方法
ぐだぐだと長くなりましたが、解決方法は以下に書いてある通りです。
https://stackoverflow.com/questions/36648128/how-to-store-custom-objects-in-dataset

僭越ながら何をしてるかを簡単に解説すると、

1. map()の中ではクラスの属性(Message.idとか)を扱うことができる
2. だから、map()を使って、(id, 元データ)のタプルを作っちゃう
3. そのタプルの"_1"(=元データのid)でjoinWithしたのち
4. 元データ取り出せばいいんじゃない？

ということをやってるっぽい。頭いいな！

# 6. ちなみに・・・
結局のところ自分のところでは最終解決に至らず・・・。
[前回の記事](https://qiita.com/ayatty/items/660157cdfb3d77cf7fc5)で書いたようにジェネリクスを使っているのだけれども、ここでどうやってEncoderを渡してあげたもんかやり方が分からない。
ジェネリクスじゃない関数（クラス決め打ち）なところは通るようになったんだけど・・・う～ん・・・。
implicitなエンコーダーが探せないんですかねえ？定義する場所が良くないのかな。その辺はまだよくわかってないのです。

# 7. まとめ
- Encoderの闇は深い
- implicitの闇はもっと深い
- が、世の中には頭いい人がいる
- ので、大抵のことに答えは見つかる
- だから、Spark始めよう！もちろんScalaでね！

とはいえ、いちいちエンコードを挟んでて実行効率はいいんだろうかこれ。今回の回避策だと余計なmap()もひとつ挟むし。